diff --git a/app/pages/Import/TaskList/TemplateModal/index.tsx b/app/pages/Import/TaskList/TemplateModal/index.tsx
index ef2e55cd..5de9538d 100644
--- a/app/pages/Import/TaskList/TemplateModal/index.tsx
+++ b/app/pages/Import/TaskList/TemplateModal/index.tsx
@@ -38,29 +38,31 @@ const TemplateModal = (props: IProps) => {
try {
const parseContent = yaml.load(content);
if(typeof parseContent === 'object') {
- const connection = parseContent.clientSettings?.connection || {};
- const address = connection.address.split(',');
- if(connection.address.startsWith('http')) {
+ const client = parseContent.client || {};
+ const address = client.address.split(',');
+ if(client.address.startsWith('http')) {
throw new Error(intl.get('import.noHttp'));
}
if(!address.includes(host)) {
throw new Error(intl.get('import.addressMatch'));
}
- if(connection.user !== username) {
+ if(client.user !== username) {
throw new Error(intl.get('import.templateMatchError', { type: 'username' }));
}
- parseContent.files?.forEach(file => {
+ parseContent.sources?.forEach(file => {
if(!files.includes(file.path)) {
throw new Error(intl.get('import.fileNotExist', { name: file.path }));
}
});
// empty props in yaml will converted to null, but its required in nebula-importer
- parseContent.files.forEach(file => {
- if(file.schema.edge) {
- file.schema.edge.props ||= [];
+ parseContent.sources.forEach(file => {
+ if(file.edges) {
+ file.edges.forEach(edge => {
+ edge.props ||= [];
+ });
}
- if(file.schema.vertex) {
- file.schema.vertex?.tags.forEach(tag => {
+ if(file.tags) {
+ file.tags.forEach(tag => {
tag.props ||= [];
});
}
diff --git a/app/utils/constant.ts b/app/utils/constant.ts
index 12749963..e8a36a1f 100644
--- a/app/utils/constant.ts
+++ b/app/utils/constant.ts
@@ -218,6 +218,5 @@ export const DEFAULT_PARTITION_NUM = 10;
export const DEFAULT_IMPORT_CONFIG = {
retry: 3,
concurrency: 10,
- channelBufferSize: 128,
batchSize: 128
};
\ No newline at end of file
diff --git a/app/utils/import.ts b/app/utils/import.ts
index aaf3514c..d3882a91 100644
--- a/app/utils/import.ts
+++ b/app/utils/import.ts
@@ -24,15 +24,11 @@ export function configToJson(payload: IConfig) {
address,
concurrency,
retry,
- channelBufferSize
} = payload;
- console.log('tagConfig', tagConfig);
- console.log('edgeConfig', edgeConfig);
const vertexToJSON = tagDataToJSON(
tagConfig,
spaceVidType,
);
- console.log('vertexToJSON', vertexToJSON);
const edgeToJSON = edgeDataToJSON(
edgeConfig,
spaceVidType,
@@ -81,22 +77,24 @@ export function edgeDataToJSON(
const edges = [{
name: handleEscape(name),
src: {
- type: vidType,
- index: srcIdIndex,
- function: srcIdFunction,
+ id: {
+ type: vidType,
+ index: srcIdIndex,
+ function: srcIdFunction,
+ }
},
dst: {
- type: vidType,
- index: dstIdIndex,
- function: dstIdFunction,
+ id: {
+ type: vidType,
+ index: dstIdIndex,
+ function: dstIdFunction,
+ }
},
- rank: { index: rank.mapping },
+ rank: typeof rank.mapping == 'number' ? { index: rank.mapping } : null,
props: edgeProps,
}];
const edgeConfig = {
- local: {
- path: file.name,
- },
+ path: file.name,
csv: {
withHeader: file.withHeader || false,
delimiter: file.delimiter
@@ -141,9 +139,7 @@ export function tagDataToJSON(
props: _props.filter(prop => prop),
}];
return {
- local: {
- path: file.name,
- },
+ path: file.name,
csv: {
withHeader: file.withHeader || false,
delimiter: file.delimiter
@@ -253,14 +249,18 @@ export const exampleJson = {
}
],
'src': {
- 'index': 1,
- 'function': null,
- 'type': 'string',
+ 'id': {
+ 'index': 1,
+ 'function': null,
+ 'type': 'string',
+ }
},
'dst': {
- 'index': 1,
- 'function': null,
- 'type': 'string',
+ 'id': {
+ 'index': 1,
+ 'function': null,
+ 'type': 'string',
+ }
},
'rank': null
}],
diff --git a/server/api/studio/internal/service/import.go b/server/api/studio/internal/service/import.go
index 390b29f3..8808dc95 100644
--- a/server/api/studio/internal/service/import.go
+++ b/server/api/studio/internal/service/import.go
@@ -14,6 +14,8 @@ import (
"sync"
"github.com/vesoft-inc/go-pkg/middleware"
+ "github.com/vesoft-inc/nebula-importer/v4/pkg/config"
+ configbase "github.com/vesoft-inc/nebula-importer/v4/pkg/config/base"
configv3 "github.com/vesoft-inc/nebula-importer/v4/pkg/config/v3"
"github.com/vesoft-inc/nebula-studio/server/api/studio/internal/service/importer"
"github.com/vesoft-inc/nebula-studio/server/api/studio/internal/svc"
@@ -61,51 +63,57 @@ func NewImportService(ctx context.Context, svcCtx *svc.ServiceContext) ImportSer
svcCtx: svcCtx,
}
}
+func updateConfig(conf configbase.Configurator, taskDir, uploadDir string) {
+ confv3 := conf.(*configv3.Config)
+ if confv3.Log == nil {
+ confv3.Log = &config.Log{}
+ confv3.Log.Files = make([]string, 0)
+ }
+
+ confv3.Log.Files = append(confv3.Log.Files, filepath.Join(taskDir, importLogName))
+ for _, source := range confv3.Sources {
+ source.SourceConfig.Local.Path = filepath.Join(uploadDir, source.SourceConfig.Local.Path)
+ }
+}
func (i *importService) CreateImportTask(req *types.CreateImportTaskRequest) (*types.CreateImportTaskData, error) {
- // 校验
- // 初始化
- // 生成目录
- // 生成配置文件
- // 启动任务
jsons, err := json.Marshal(req.Config)
if err != nil {
return nil, ecode.WithErrorMessage(ecode.ErrParam, err)
}
- conf := configv3.Config{}
- err = json.Unmarshal(jsons, &conf)
- if err != nil {
- return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
- }
- // init task
- auth := i.ctx.Value(auth.CtxKeyUserInfo{}).(*auth.AuthData)
- host := auth.Address + ":" + strconv.Itoa(auth.Port)
- taskMgr := importer.GetTaskMgr()
- task, taskID, err := taskMgr.NewTask(host, auth.Username, req.Name, conf)
+
+ conf, err := config.FromBytes(jsons)
+
if err != nil {
- logx.Errorf("init task fail", err)
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
+
// create task dir
- taskDir, err := importer.CreateTaskDir(i.svcCtx.Config.File.TasksDir, taskID)
+ taskDir, err := importer.CreateNewTaskDir(i.svcCtx.Config.File.TasksDir)
if err != nil {
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
// create config file
- if err := importer.CreateConfigFile(taskDir, conf); err != nil {
+ if err := importer.CreateConfigFile(taskDir, jsons); err != nil {
+ return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
+ }
+ // modify source file path & add log config
+ updateConfig(conf, taskDir, i.svcCtx.Config.File.UploadDir)
+
+ // init task in db
+ auth := i.ctx.Value(auth.CtxKeyUserInfo{}).(*auth.AuthData)
+ host := auth.Address + ":" + strconv.Itoa(auth.Port)
+ taskMgr := importer.GetTaskMgr()
+ task, taskID, err := taskMgr.NewTask(host, auth.Username, req.Name, conf)
+ if err != nil {
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
// start import
if err = importer.StartImport(taskID); err != nil {
- // task err: import task not start err
task.TaskInfo.TaskStatus = importer.StatusAborted.String()
- err1 := importer.GetTaskMgr().AbortTask(taskID)
- if err != nil {
- logx.Errorf("finish task fail", err1)
- }
- logx.Errorf(fmt.Sprintf("Failed to start a import task: `%s`, task result: `%v`", taskID, err))
+ importer.GetTaskMgr().AbortTask(taskID)
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
@@ -113,25 +121,17 @@ func (i *importService) CreateImportTask(req *types.CreateImportTaskRequest) (*t
muTaskId.Lock()
taskIDBytes, err := ioutil.ReadFile(i.svcCtx.Config.File.TaskIdPath)
if err != nil {
- logx.Errorf("read taskId file error", err)
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
taskIdJSON := make(map[string]bool)
if len(taskIDBytes) != 0 {
if err := json.Unmarshal(taskIDBytes, &taskIdJSON); err != nil {
- logx.Errorf("read taskId file error", err)
return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
}
- taskIdJSON[taskID] = true
- bytes, err := json.Marshal(taskIdJSON)
- if err != nil {
- logx.Errorf("read taskId file error", err)
- }
- err = ioutil.WriteFile(i.svcCtx.Config.File.TaskIdPath, bytes, 777)
- if err != nil {
- logx.Errorf("write taskId file error", err)
- }
+ taskIdJSON[strconv.Itoa(taskID)] = true
+ bytes, _ := json.Marshal(taskIdJSON)
+ ioutil.WriteFile(i.svcCtx.Config.File.TaskIdPath, bytes, 777)
defer muTaskId.Unlock()
return &types.CreateImportTaskData{
@@ -156,7 +156,7 @@ func (i *importService) DownloadConfig(req *types.DownloadConfigsRequest) error
return ecode.WithInternalServer(fmt.Errorf("unset KeepResponse Writer"))
}
- configPath := filepath.Join(i.svcCtx.Config.File.TasksDir, req.Id, "config.yaml")
+ configPath := filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(req.Id), "config.yaml")
httpResp.Header().Set("Content-Type", "application/octet-stream")
httpResp.Header().Set("Content-Disposition", "attachment;filename="+filepath.Base(configPath))
http.ServeFile(httpResp, httpReq, configPath)
@@ -180,9 +180,9 @@ func (i *importService) DownloadLogs(req *types.DownloadLogsRequest) error {
filename := req.Name
path := ""
if filename == importLogName {
- path = filepath.Join(i.svcCtx.Config.File.TasksDir, id, filename)
+ path = filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(id), filename)
} else {
- path = filepath.Join(i.svcCtx.Config.File.TasksDir, id, "err", filename)
+ path = filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(id), "err", filename)
}
httpResp.Header().Set("Content-Type", "application/octet-stream")
@@ -211,31 +211,32 @@ func (i *importService) GetManyImportTask(req *types.GetManyImportTaskRequest) (
// GetImportTaskLogNames :Get all log file's name of a task
func (i *importService) GetImportTaskLogNames(req *types.GetImportTaskLogNamesRequest) (*types.GetImportTaskLogNamesData, error) {
- id := req.Id
+ // TODO err log will be support in next importer version
+ // id := req.Id
- errLogDir := filepath.Join(i.svcCtx.Config.File.TasksDir, id, "err")
- fileInfos, err := ioutil.ReadDir(errLogDir)
- if err != nil {
- return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
- }
+ // errLogDir := filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(id), "err")
+ // fileInfos, err := ioutil.ReadDir(errLogDir)
+ // if err != nil {
+ // return nil, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
+ // }
data := &types.GetImportTaskLogNamesData{
Names: []string{},
}
data.Names = append(data.Names, importLogName)
- for _, fileInfo := range fileInfos {
- name := fileInfo.Name()
- data.Names = append(data.Names, name)
- }
+ // for _, fileInfo := range fileInfos {
+ // name := fileInfo.Name()
+ // data.Names = append(data.Names, name)
+ // }
return data, nil
}
func (i *importService) GetManyImportTaskLog(req *types.GetManyImportTaskLogRequest) (*types.GetManyImportTaskLogData, error) {
path := ""
if req.File == importLogName {
- path = filepath.Join(i.svcCtx.Config.File.TasksDir, req.Id, req.File)
+ path = filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(req.Id), req.File)
} else {
- path = filepath.Join(i.svcCtx.Config.File.TasksDir, req.Id, errContentDir, req.File)
+ path = filepath.Join(i.svcCtx.Config.File.TasksDir, strconv.Itoa(req.Id), errContentDir, req.File)
}
lines, err := readFileLines(path, req.Offset, req.Limit)
if err != nil {
@@ -260,7 +261,7 @@ func (i *importService) GetManyImportTaskLog(req *types.GetManyImportTaskLogRequ
data := &types.GetManyImportTaskLogData{
Logs: lines,
}
- if len(lines) == 0 && taskIdJSON[req.Id] {
+ if len(lines) == 0 && taskIdJSON[strconv.Itoa(req.Id)] {
return data, nil
}
if len(lines) == 0 {
@@ -277,20 +278,6 @@ func (i *importService) GetWorkingDir() (*types.GetWorkingDirResult, error) {
}, nil
}
-// func validClientParams(conf *importconfig.YAMLConfig) error {
-// if conf.NebulaClientSettings.Connection == nil ||
-// conf.NebulaClientSettings.Connection.Address == nil ||
-// *conf.NebulaClientSettings.Connection.Address == "" ||
-// conf.NebulaClientSettings.Connection.User == nil ||
-// *conf.NebulaClientSettings.Connection.User == "" ||
-// conf.NebulaClientSettings.Space == nil ||
-// *conf.NebulaClientSettings.Space == "" {
-// return ecode.WithCode(ecode.ErrParam, nil)
-// }
-
-// return nil
-// }
-
func readFileLines(path string, offset int64, limit int64) ([]string, error) {
file, err := os.Open(path)
if err != nil {
diff --git a/server/api/studio/internal/service/importer/importer.go b/server/api/studio/internal/service/importer/importer.go
index 759221c2..75e1e877 100644
--- a/server/api/studio/internal/service/importer/importer.go
+++ b/server/api/studio/internal/service/importer/importer.go
@@ -2,14 +2,11 @@ package importer
import (
"errors"
- "fmt"
"regexp"
- "strconv"
"time"
"github.com/vesoft-inc/nebula-studio/server/api/studio/internal/types"
"github.com/vesoft-inc/nebula-studio/server/api/studio/pkg/ecode"
- "github.com/zeromicro/go-zero/core/logx"
)
type ImportResult struct {
@@ -22,19 +19,15 @@ type ImportResult struct {
}
}
-func StartImport(taskID string) (err error) {
+func StartImport(taskID int) (err error) {
task, _ := GetTaskMgr().GetTask(taskID)
signal := make(chan struct{}, 1)
abort := func() {
task.TaskInfo.TaskStatus = StatusAborted.String()
task.TaskInfo.TaskMessage = err.Error()
- err = GetTaskMgr().AbortTask(taskID)
- if err != nil {
- logx.Errorf("start task fail, %v", err)
- }
+ GetTaskMgr().AbortTask(taskID)
signal <- struct{}{}
- return
}
go func() {
@@ -43,73 +36,60 @@ func StartImport(taskID string) (err error) {
for {
select {
case <-ticker.C:
- err := GetTaskMgr().UpdateTaskInfo(taskID)
- if err != nil {
- logx.Errorf(fmt.Sprintf("UpdateTaskInfo fail, id : %s", taskID), err)
- }
+ GetTaskMgr().UpdateTaskInfo(taskID)
case <-signal:
return
}
}
}()
go func() {
- mgr := task.Client.Manager
cfg := task.Client.Cfg
if err = cfg.Build(); err != nil {
abort()
+ return
}
+ mgr := cfg.GetManager()
+ logger := cfg.GetLogger()
+ task.Client.Manager = mgr
+ task.Client.Logger = logger
+
if err = mgr.Start(); err != nil {
abort()
+ return
}
err = mgr.Wait()
if err != nil {
task.TaskInfo.TaskStatus = StatusAborted.String()
task.TaskInfo.TaskMessage = err.Error()
- err = GetTaskMgr().AbortTask(taskID)
- if err != nil {
- logx.Errorf("finish task fail, %v", err)
- }
+ GetTaskMgr().AbortTask(taskID)
return
}
task.TaskInfo.TaskStatus = StatusFinished.String()
- err = GetTaskMgr().FinishTask(taskID)
- if err != nil {
- logx.Errorf("finish task fail, %v", err)
- }
+ GetTaskMgr().FinishTask(taskID)
signal <- struct{}{}
}()
return nil
}
-func DeleteImportTask(tasksDir, taskID, address, username string) error {
- if id, err := strconv.Atoi(taskID); err != nil {
- logx.Errorf(fmt.Sprintf("stop task fail, id : %s", taskID), err)
- return errors.New("task not existed")
- } else {
- _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(id, address, username)
- if err != nil {
- logx.Errorf(fmt.Sprintf("stop task fail, id : %s", taskID), err)
- return errors.New("task not existed")
- }
+func DeleteImportTask(tasksDir string, taskID int, address, username string) error {
+ _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(taskID, address, username)
+ if err != nil {
+ return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
- err := GetTaskMgr().DelTask(tasksDir, taskID)
+ err = GetTaskMgr().DelTask(tasksDir, taskID)
if err != nil {
- return fmt.Errorf("task del fail, %s", err.Error())
+ return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
return nil
}
-func GetImportTask(tasksDir, taskID, address, username string) (*types.GetImportTaskData, error) {
+func GetImportTask(tasksDir string, taskID int, address, username string) (*types.GetImportTaskData, error) {
task := Task{}
result := &types.GetImportTaskData{}
- if id, err := strconv.Atoi(taskID); err != nil {
+ _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(taskID, address, username)
+ if err != nil {
return nil, errors.New("task not existed")
- } else {
- _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(id, address, username)
- if err != nil {
- return nil, errors.New("task not existed")
- }
}
if t, ok := GetTaskMgr().GetTask(taskID); ok {
@@ -119,7 +99,7 @@ func GetImportTask(tasksDir, taskID, address, username string) (*types.GetImport
return nil, err
}
stats := task.TaskInfo.Stats
- result.Id = strconv.Itoa(t.TaskInfo.ID)
+ result.Id = t.TaskInfo.ID
result.Status = task.TaskInfo.TaskStatus
result.Message = task.TaskInfo.TaskMessage
result.CreateTime = task.TaskInfo.CreateTime.UnixMilli()
@@ -164,7 +144,7 @@ func GetManyImportTask(tasksDir, address, username string, pageIndex, pageSize i
}
stats := t.Stats
data := types.GetImportTaskData{
- Id: strconv.Itoa(t.ID),
+ Id: t.ID,
Status: t.TaskStatus,
Message: t.TaskMessage,
CreateTime: t.CreateTime.UnixMilli(),
@@ -194,21 +174,14 @@ func GetManyImportTask(tasksDir, address, username string, pageIndex, pageSize i
return result, nil
}
-func StopImportTask(taskID, address, username string) error {
- if id, err := strconv.Atoi(taskID); err != nil {
- logx.Errorf(fmt.Sprintf("stop task fail, id : %s", taskID), err)
- return errors.New("task not existed")
- } else {
- _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(id, address, username)
- if err != nil {
- logx.Errorf(fmt.Sprintf("stop task fail, id : %s", taskID), err)
- return errors.New("task not existed")
- }
+func StopImportTask(taskID int, address, username string) error {
+ _, err := taskmgr.db.FindTaskInfoByIdAndAddresssAndUser(taskID, address, username)
+ if err != nil {
+ return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
- err := GetTaskMgr().StopTask(taskID)
+ err = GetTaskMgr().StopTask(taskID)
if err != nil {
- logx.Errorf(fmt.Sprintf("stop task fail, id : %s", taskID), err)
return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
} else {
return nil
diff --git a/server/api/studio/internal/service/importer/task.go b/server/api/studio/internal/service/importer/task.go
index 27fa61c7..72363abe 100644
--- a/server/api/studio/internal/service/importer/task.go
+++ b/server/api/studio/internal/service/importer/task.go
@@ -14,7 +14,7 @@ type Client struct {
Manager manager.Manager `json:"manager,omitempty"`
}
type Task struct {
- Client Client `json:"client,omitempty"`
+ Client *Client `json:"client,omitempty"`
TaskInfo *db.TaskInfo `json:"task_info,omitempty"`
}
diff --git a/server/api/studio/internal/service/importer/taskmgr.go b/server/api/studio/internal/service/importer/taskmgr.go
index 3fe9eead..6d39687d 100644
--- a/server/api/studio/internal/service/importer/taskmgr.go
+++ b/server/api/studio/internal/service/importer/taskmgr.go
@@ -1,9 +1,7 @@
package importer
import (
- "encoding/json"
"errors"
- "fmt"
"os"
"path/filepath"
"strconv"
@@ -14,9 +12,7 @@ import (
db "github.com/vesoft-inc/nebula-studio/server/api/studio/internal/model"
"github.com/vesoft-inc/nebula-studio/server/api/studio/pkg/ecode"
"github.com/vesoft-inc/nebula-studio/server/api/studio/pkg/utils"
- "github.com/zeromicro/go-zero/core/logx"
- "go.uber.org/zap"
- "gopkg.in/yaml.v2"
+ "gopkg.in/yaml.v3"
_ "github.com/mattn/go-sqlite3"
)
@@ -35,80 +31,69 @@ type TaskMgr struct {
db *TaskDb
}
-func createTaskClient(conf configv3.Config) (client Client, err error) {
- jsons, err := json.Marshal(conf)
+func CreateNewTaskDir(rootDir string) (string, error) {
+ taskId, err := GetTaskMgr().NewTaskID()
if err != nil {
- return Client{}, ecode.WithErrorMessage(ecode.ErrParam, err)
- }
- cfg, err := importConfig.FromBytes(jsons)
- if err != nil {
- return Client{}, ecode.WithErrorMessage(ecode.ErrInternalServer, err)
- }
- client = Client{
- Cfg: cfg,
- Manager: cfg.GetManager(),
- Logger: cfg.GetLogger(),
+ return "", ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
- return client, nil
-}
-
-func CreateTaskDir(rootDir string, id string) (string, error) {
- taskDir := filepath.Join(rootDir, id)
+ taskDir := filepath.Join(rootDir, strconv.Itoa(taskId))
if err := utils.CreateDir(taskDir); err != nil {
return "", ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
return taskDir, nil
}
-func CreateConfigFile(taskdir string, config configv3.Config) error {
+func CreateConfigFile(taskdir string, cfgBytes []byte) error {
fileName := "config.yaml"
path := filepath.Join(taskdir, fileName)
+ config, _ := importConfig.FromBytes(cfgBytes)
+ confv3 := config.(*configv3.Config)
+
// erase user information
- _config := config
- _config.Client.User = ""
- _config.Client.Password = ""
+ _config := confv3
+ _config.Client.User = "YOUR_NEBULA_NAME"
+ _config.Client.Password = "YOUR_NEBULA_PASSWORD"
_config.Client.Address = ""
-
// TODO hide data source access key and so on
-
- outYaml, err := yaml.Marshal(config)
+ outYaml, err := yaml.Marshal(confv3)
if err != nil {
return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
if err := os.WriteFile(path, outYaml, 0o644); err != nil {
- zap.L().Warn("write"+path+"file error", zap.Error(err))
return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
return nil
}
-func (mgr *TaskMgr) NewTask(host string, user string, taskName string, conf configv3.Config) (*Task, string, error) {
+func (mgr *TaskMgr) NewTask(host string, user string, taskName string, cfg importConfig.Configurator) (*Task, int, error) {
mux.Lock()
defer mux.Unlock()
- // init importer client
- client, err := createTaskClient(conf)
- if err != nil {
- return nil, "", err
- }
+ confv3 := cfg.(*configv3.Config)
+
// init task db
taskInfo := &db.TaskInfo{
Name: taskName,
Address: host,
- Space: conf.Manager.GraphName,
+ Space: confv3.Manager.GraphName,
TaskStatus: StatusProcessing.String(),
- ImportAddress: conf.Client.Address,
+ ImportAddress: confv3.Client.Address,
User: user,
}
+
+ if err := mgr.db.InsertTaskInfo(taskInfo); err != nil {
+ return nil, 0, err
+ }
+
task := &Task{
- Client: client,
+ Client: &Client{
+ Cfg: cfg,
+ Manager: nil,
+ Logger: nil,
+ },
TaskInfo: taskInfo,
}
- if err := mgr.db.InsertTaskInfo(task.TaskInfo); err != nil {
- return nil, "", err
- }
- // create task dir
- id := string(taskInfo.ID)
+ id := int(taskInfo.ID)
mgr.PutTask(id, task)
return task, id, nil
}
@@ -117,10 +102,18 @@ func GetTaskMgr() *TaskMgr {
return taskmgr
}
+func (mgr *TaskMgr) NewTaskID() (int, error) {
+ tid, err := mgr.db.LastId()
+ if err != nil {
+ return 0, err
+ }
+ return tid + 1, nil
+}
+
/*
GetTask get task from map and local sql
*/
-func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool) {
+func (mgr *TaskMgr) GetTask(taskID int) (*Task, bool) {
if task, ok := mgr.getTaskFromMap(taskID); ok {
return task, true
}
@@ -135,7 +128,7 @@ func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool) {
/*
PutTask put task into tasks map
*/
-func (mgr *TaskMgr) PutTask(taskID string, task *Task) {
+func (mgr *TaskMgr) PutTask(taskID int, task *Task) {
mgr.tasks.Store(taskID, task)
}
@@ -143,7 +136,7 @@ func (mgr *TaskMgr) PutTask(taskID string, task *Task) {
FinishTask will query task stats, delete task in the map
and update the taskInfo in local sql
*/
-func (mgr *TaskMgr) FinishTask(taskID string) (err error) {
+func (mgr *TaskMgr) FinishTask(taskID int) (err error) {
task, ok := mgr.getTaskFromMap(taskID)
if !ok {
return
@@ -159,7 +152,7 @@ func (mgr *TaskMgr) FinishTask(taskID string) (err error) {
return
}
-func (mgr *TaskMgr) AbortTask(taskID string) (err error) {
+func (mgr *TaskMgr) AbortTask(taskID int) (err error) {
task, ok := mgr.getTaskFromMap(taskID)
if !ok {
return
@@ -172,19 +165,19 @@ func (mgr *TaskMgr) AbortTask(taskID string) (err error) {
return
}
-func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error {
+func (mgr *TaskMgr) DelTask(tasksDir string, taskID int) error {
_, ok := mgr.getTaskFromMap(taskID)
if ok {
mgr.tasks.Delete(taskID)
}
- id, err := strconv.Atoi(taskID)
- if err != nil {
- return errors.New("taskID is wrong")
- }
- if err = mgr.db.DelTaskInfo(id); err != nil {
+ // id, err := strconv.Atoi(taskID)
+ // if err != nil {
+ // return errors.New("taskID is wrong")
+ // }
+ if err := mgr.db.DelTaskInfo(taskID); err != nil {
return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
- taskDir := filepath.Join(tasksDir, taskID)
+ taskDir := filepath.Join(tasksDir, strconv.Itoa(taskID))
return os.RemoveAll(taskDir)
}
@@ -192,7 +185,7 @@ func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error {
UpdateTaskInfo will query task stats, update task in the map
and update the taskInfo in local sql
*/
-func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error {
+func (mgr *TaskMgr) UpdateTaskInfo(taskID int) error {
task, ok := mgr.getTaskFromMap(taskID)
if !ok {
return nil
@@ -207,7 +200,7 @@ func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error {
StopTask will change the task status to `StatusStoped`,
and then call FinishTask
*/
-func (mgr *TaskMgr) StopTask(taskID string) error {
+func (mgr *TaskMgr) StopTask(taskID int) error {
if task, ok := mgr.getTaskFromMap(taskID); ok {
manager := task.Client.Manager
err := manager.Stop()
@@ -216,7 +209,6 @@ func (mgr *TaskMgr) StopTask(taskID string) error {
}
task.TaskInfo.TaskStatus = StatusStoped.String()
if err := mgr.FinishTask(taskID); err != nil {
- logx.Alert(fmt.Sprintf("finish task fail: %s", err))
return ecode.WithErrorMessage(ecode.ErrInternalServer, err)
}
return nil
@@ -239,14 +231,14 @@ func (mgr *TaskMgr) GetAllTaskIDs(address, username string) ([]string, error) {
return ids, nil
}
-func (mgr *TaskMgr) getTaskFromMap(taskID string) (*Task, bool) {
+func (mgr *TaskMgr) getTaskFromMap(taskID int) (*Task, bool) {
if task, ok := mgr.tasks.Load(taskID); ok {
return task.(*Task), true
}
return nil, false
}
-func (mgr *TaskMgr) getTaskFromSQL(taskID string) *Task {
+func (mgr *TaskMgr) getTaskFromSQL(taskID int) *Task {
taskInfo := new(db.TaskInfo)
mgr.db.First(taskInfo, taskID)
task := new(Task)
diff --git a/server/api/studio/internal/types/types.go b/server/api/studio/internal/types/types.go
index 2840ab9d..a52f1e2a 100644
--- a/server/api/studio/internal/types/types.go
+++ b/server/api/studio/internal/types/types.go
@@ -55,41 +55,45 @@ type ImportTaskCSV struct {
}
type NodeId struct {
- Name string `json:"name" validate:"required"`
+ Name string `json:"name,optional"`
Type string `json:"type" validate:"required"`
Index int64 `json:"index" validate:"required"`
- ConcatItems []interface{} `json:"concatItems,optional"`
- Function string `json:"function,optional"`
+ ConcatItems []interface{} `json:"concatItems,optional,omitempty"`
+ Function string `json:"function,optional,omitempty"`
}
type Tag struct {
Name string `json:"name" validate:"required"`
ID NodeId `json:"id" validate:"required"`
Props []Prop `json:"props" validate:"required"`
- IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional"`
+ IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional,omitempty"`
}
type Edge struct {
- Name string `json:"name" validate:"required"`
- Src NodeId `json:"src" validate:"required"`
- Dst NodeId `json:"dst" validate:"required"`
- Props []Prop `json:"props" validate:"required"`
- Rank EdgeRank `json:"rank, optional"`
- IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional"`
+ Name string `json:"name" validate:"required"`
+ Src EdgeNodeRef `json:"src" validate:"required"`
+ Dst EdgeNodeRef `json:"dst" validate:"required"`
+ Props []Prop `json:"props" validate:"required"`
+ Rank *EdgeRank `json:"rank,optional,omitempty"`
+ IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional,omitempty"`
+}
+
+type EdgeNodeRef struct {
+ ID NodeId `json:"id" validate:"required"`
}
type Prop struct {
Name string `json:"name" validate:"required"`
Type string `json:"type" validate:"required"`
- Index int64 `json:"index, optional"`
- Nullable bool `json:"nullable, optional"`
- NullValue string `json:"nullValue, optional"`
- AlternativeIndices []int64 `json:"alternativeIndices, optional"`
- DefaultValue string `json:"defaultValue, optional"`
+ Index int64 `json:"index,optional,omitempty"`
+ Nullable bool `json:"nullable,optional,omitempty"`
+ NullValue string `json:"nullValue,optional,omitempty"`
+ AlternativeIndices []int64 `json:"alternativeIndices,optional,omitempty"`
+ DefaultValue string `json:"defaultValue,optional,omitempty"`
}
type EdgeRank struct {
- Index *int64 `json:"index, optional"`
+ Index *int64 `json:"index,optional,omitempty"`
}
type S3Config struct {
@@ -121,34 +125,47 @@ type ImportTaskConfig struct {
Client Client `json:"client" validate:"required"`
Manager Manager `json:"manager" validate:"required"`
Sources []Sources `json:"sources" validate:"required"`
+ Log *Log `json:"log,omitempty,optional"`
}
type Client struct {
- Version string `json:"version,omitempty" validate:"required"`
- Address string `json:"address,omitempty" validate:"required"`
- User string `json:"user,omitempty" validate:"required"`
- Password string `json:"password,omitempty" validate:"required"`
- ConcurrencyPerAddress int `json:"concurrencyPerAddress,optional"`
- ReconnectInitialInterval string `json:"reconnectInitialInterval,optional"`
- Retry int `json:"retry,optional"`
- RetryInitialInterval string `json:"retryInitialInterval,optional"`
+ Version string `json:"version,omitempty" validate:"required"`
+ Address string `json:"address,omitempty" validate:"required"`
+ User string `json:"user,omitempty" validate:"required"`
+ Password string `json:"password,omitempty" validate:"required"`
+ ConcurrencyPerAddress int `json:"concurrencyPerAddress,optional"`
+ ReconnectInitialInterval *string `json:"reconnectInitialInterval,optional,omitempty"`
+ Retry int `json:"retry,optional"`
+ RetryInitialInterval *string `json:"retryInitialInterval,optional,omitempty"`
}
type Manager struct {
- SpaceName string `json:"spaceName,omitempty" validate:"required"`
- Batch int `json:"batch,omitempty, optional"`
- ReaderConcurrency int `json:"readerConcurrency,omitempty, optional"`
- ImporterConcurrency int `json:"importerConcurrency,omitempty, optional"`
- StatsInterval string `json:"statsInterval,omitempty, optional"`
+ SpaceName string `json:"spaceName,omitempty" validate:"required"`
+ Batch int `json:"batch,omitempty,optional"`
+ ReaderConcurrency int `json:"readerConcurrency,omitempty,optional"`
+ ImporterConcurrency int `json:"importerConcurrency,omitempty,optional"`
+ StatsInterval *string `json:"statsInterval,omitempty,optional"`
}
type Sources struct {
CSV ImportTaskCSV `json:"csv" validate:"required"`
- Local LocalConfig `json:"local, optional"`
- S3 S3Config `json:"s3, optional"`
- SFTP SFTPConfig `json:"sftpConfig, optional"`
- Tags []Tag `json:"tags, optional"`
- Edges []Edge `json:"edges, optional"`
+ Path string `json:"path,optional,omitempty"`
+ S3 *S3Config `json:"s3,optional,omitempty"`
+ SFTP *SFTPConfig `json:"sftpConfig,optional,omitempty"`
+ Tags []Tag `json:"tags,optional"`
+ Edges []Edge `json:"edges,optional"`
+}
+
+type Log struct {
+ Level *string `json:"level,omitempty,optional"`
+ Console *bool `json:"console,omitempty,optional"`
+ Files []string `json:"files,omitempty,optional"`
+ Fields []LogField `json:"fields,omitempty,optional"`
+}
+
+type LogField struct {
+ Key string `json:"key"`
+ Value interface{} `json:"value"`
}
type CreateImportTaskRequest struct {
@@ -157,15 +174,15 @@ type CreateImportTaskRequest struct {
}
type CreateImportTaskData struct {
- Id string `json:"id"`
+ Id int `json:"id"`
}
type GetImportTaskRequest struct {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
}
type GetImportTaskData struct {
- Id string `json:"id"`
+ Id int `json:"id"`
Name string `json:"name"`
User string `json:"user"`
Address string `json:"address"`
@@ -202,7 +219,7 @@ type GetManyImportTaskData struct {
}
type GetManyImportTaskLogRequest struct {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
File string `form:"file" validate:"required"`
Offset int64 `form:"offset" validate:"min=0"`
Limit int64 `form:"limit" validate:"min=1"`
@@ -213,7 +230,7 @@ type GetManyImportTaskLogData struct {
}
type GetImportTaskLogNamesRequest struct {
- Id string `path:"id" validate:"required""`
+ Id int `path:"id" validate:"required"`
}
type GetImportTaskLogNamesData struct {
@@ -221,20 +238,20 @@ type GetImportTaskLogNamesData struct {
}
type DeleteImportTaskRequest struct {
- Id string `path:"id"`
+ Id int `path:"id"`
}
type StopImportTaskRequest struct {
- Id string `path:"id"`
+ Id int `path:"id"`
}
type DownloadLogsRequest struct {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
Name string `form:"name" validate:"required"`
}
type DownloadConfigsRequest struct {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
}
type GetWorkingDirResult struct {
diff --git a/server/api/studio/restapi/import.api b/server/api/studio/restapi/import.api
index 1f3f11cf..3480ab0a 100644
--- a/server/api/studio/restapi/import.api
+++ b/server/api/studio/restapi/import.api
@@ -11,37 +11,41 @@ type (
Name string `json:"name,optional"`
Type string `json:"type" validate:"required"`
Index int64 `json:"index" validate:"required"`
- ConcatItems []interface{} `json:"concatItems,optional"`
- Function string `json:"function,optional"`
+ ConcatItems []interface{} `json:"concatItems,optional,omitempty"`
+ Function string `json:"function,optional,omitempty"`
}
Tag {
Name string `json:"name" validate:"required"`
ID NodeId `json:"id" validate:"required"`
Props []Prop `json:"props" validate:"required"`
- IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional"`
+ IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional,omitempty"`
}
Edge {
- Name string `json:"name" validate:"required"`
- Src NodeId `json:"src" validate:"required"`
- Dst NodeId `json:"dst" validate:"required"`
- Props []Prop `json:"props" validate:"required"`
- Rank EdgeRank `json:"rank, optional"`
- IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional"`
+ Name string `json:"name" validate:"required"`
+ Src EdgeNodeRef `json:"src" validate:"required"`
+ Dst EdgeNodeRef `json:"dst" validate:"required"`
+ Props []Prop `json:"props" validate:"required"`
+ Rank *EdgeRank `json:"rank,optional,omitempty"`
+ IgnoreExistedIndex bool `json:"ignoreExistedIndex,optional,omitempty"`
+ }
+
+ EdgeNodeRef {
+ ID NodeId `json:"id" validate:"required"`
}
Prop {
Name string `json:"name" validate:"required"`
Type string `json:"type" validate:"required"`
- Index int64 `json:"index, optional"`
- Nullable bool `json:"nullable, optional"`
- NullValue string `json:"nullValue, optional"`
- AlternativeIndices []int64 `json:"alternativeIndices, optional"`
- DefaultValue string `json:"defaultValue, optional"`
+ Index int64 `json:"index,optional,omitempty"`
+ Nullable bool `json:"nullable,optional,omitempty"`
+ NullValue string `json:"nullValue,optional,omitempty"`
+ AlternativeIndices []int64 `json:"alternativeIndices,optional,omitempty"`
+ DefaultValue string `json:"defaultValue,optional,omitempty"`
}
EdgeRank {
- Index *int64 `json:"index, optional"`
+ Index *int64 `json:"index,optional,omitempty"`
}
S3Config {
@@ -73,34 +77,47 @@ type (
Client Client `json:"client" validate:"required"`
Manager Manager `json:"manager" validate:"required"`
Sources []Sources `json:"sources" validate:"required"`
+ Log *Log `json:"log,omitempty,optional"`
}
Client {
- Version string `json:"version,omitempty" validate:"required"`
- Address string `json:"address,omitempty" validate:"required"`
- User string `json:"user,omitempty" validate:"required"`
- Password string `json:"password,omitempty" validate:"required"`
- ConcurrencyPerAddress int `json:"concurrencyPerAddress,optional"`
- ReconnectInitialInterval string `json:"reconnectInitialInterval,optional"`
- Retry int `json:"retry,optional"`
- RetryInitialInterval string `json:"retryInitialInterval,optional"`
+ Version string `json:"version,omitempty" validate:"required"`
+ Address string `json:"address,omitempty" validate:"required"`
+ User string `json:"user,omitempty" validate:"required"`
+ Password string `json:"password,omitempty" validate:"required"`
+ ConcurrencyPerAddress int `json:"concurrencyPerAddress,optional"`
+ ReconnectInitialInterval *string `json:"reconnectInitialInterval,optional,omitempty"`
+ Retry int `json:"retry,optional"`
+ RetryInitialInterval *string `json:"retryInitialInterval,optional,omitempty"`
}
Manager {
- SpaceName string `json:"spaceName,omitempty" validate:"required"`
- Batch int `json:"batch,omitempty, optional"`
- ReaderConcurrency int `json:"readerConcurrency,omitempty, optional"`
- ImporterConcurrency int `json:"importerConcurrency,omitempty, optional"`
- StatsInterval string `json:"statsInterval,omitempty, optional"`
+ SpaceName string `json:"spaceName,omitempty" validate:"required"`
+ Batch int `json:"batch,omitempty,optional"`
+ ReaderConcurrency int `json:"readerConcurrency,omitempty,optional"`
+ ImporterConcurrency int `json:"importerConcurrency,omitempty,optional"`
+ StatsInterval *string `json:"statsInterval,omitempty,optional"`
}
Sources {
CSV ImportTaskCSV `json:"csv" validate:"required"`
- Local LocalConfig `json:"local, optional"`
- S3 S3Config `json:"s3, optional"`
- SFTP SFTPConfig `json:"sftpConfig, optional"`
- Tags []Tag `json:"tags, optional"`
- Edges []Edge `json:"edges, optional"`
+ Path string `json:"path,optional,omitempty"`
+ S3 *S3Config `json:"s3,optional,omitempty"`
+ SFTP *SFTPConfig `json:"sftpConfig,optional,omitempty"`
+ Tags []Tag `json:"tags,optional"`
+ Edges []Edge `json:"edges,optional"`
+ }
+
+ Log {
+ Level *string `json:"level,omitempty,optional"`
+ Console *bool `json:"console,omitempty,optional"`
+ Files []string `json:"files,omitempty,optional"`
+ Fields []LogField `json:"fields,omitempty,optional"`
+ }
+
+ LogField {
+ Key string `json:"key"`
+ Value interface{} `json:"value"`
}
CreateImportTaskRequest {
@@ -109,15 +126,15 @@ type (
}
CreateImportTaskData {
- Id string `json:"id"`
+ Id int `json:"id"`
}
GetImportTaskRequest {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
}
GetImportTaskData {
- Id string `json:"id"`
+ Id int `json:"id"`
Name string `json:"name"`
User string `json:"user"`
Address string `json:"address"`
@@ -154,7 +171,7 @@ type (
}
GetManyImportTaskLogRequest {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
File string `form:"file" validate:"required"`
Offset int64 `form:"offset" validate:"min=0"`
Limit int64 `form:"limit" validate:"min=1"`
@@ -165,7 +182,7 @@ type (
}
GetImportTaskLogNamesRequest {
- Id string `path:"id" validate:"required""`
+ Id int `path:"id" validate:"required"`
}
GetImportTaskLogNamesData {
@@ -173,20 +190,20 @@ type (
}
DeleteImportTaskRequest {
- Id string `path:"id"`
+ Id int `path:"id"`
}
StopImportTaskRequest {
- Id string `path:"id"`
+ Id int `path:"id"`
}
DownloadLogsRequest {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
Name string `form:"name" validate:"required"`
}
DownloadConfigsRequest {
- Id string `path:"id" validate:"required"`
+ Id int `path:"id" validate:"required"`
}
GetWorkingDirResult {
diff --git a/server/go.mod b/server/go.mod
index 6e902392..1161982a 100644
--- a/server/go.mod
+++ b/server/go.mod
@@ -17,6 +17,7 @@ require (
github.com/pkg/sftp v1.13.5
github.com/stretchr/testify v1.8.0
github.com/vesoft-inc/nebula-go/v3 v3.4.0
+ gopkg.in/yaml.v3 v3.0.1
)
require (
@@ -51,10 +52,9 @@ require (
go.uber.org/atomic v1.9.0 // indirect
go.uber.org/multierr v1.8.0 // indirect
golang.org/x/time v0.3.0 // indirect
- gopkg.in/yaml.v3 v3.0.1 // indirect
)
-replace github.com/vesoft-inc/nebula-importer/v4 => github.com/veezhang/nebula-importer/v4 v4.0.0-20230302074940-d97c824a6225
+replace github.com/vesoft-inc/nebula-importer/v4 => github.com/veezhang/nebula-importer/v4 v4.0.0-20230308081941-4dad7d3f5cb1
require (
github.com/axgle/mahonia v0.0.0-20180208002826-3358181d7394
@@ -93,6 +93,6 @@ require (
golang.org/x/text v0.6.0 // indirect
google.golang.org/grpc v1.46.0 // indirect
google.golang.org/protobuf v1.28.1 // indirect
- gopkg.in/yaml.v2 v2.4.0
+ gopkg.in/yaml.v2 v2.4.0 // indirect
gorm.io/driver/sqlite v1.3.2
)
diff --git a/server/go.sum b/server/go.sum
index 5ed4ade6..aa480a56 100644
--- a/server/go.sum
+++ b/server/go.sum
@@ -451,8 +451,8 @@ github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhV
github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
-github.com/veezhang/nebula-importer/v4 v4.0.0-20230302074940-d97c824a6225 h1:BZv/ScJIgYQSeVVJWPDgJKrEhn5TQJ2ZngBpDvGbA9o=
-github.com/veezhang/nebula-importer/v4 v4.0.0-20230302074940-d97c824a6225/go.mod h1:6Z0GAlZdHWMQCyLkq377Li6ibIHkcwkzmBTEwIorAoU=
+github.com/veezhang/nebula-importer/v4 v4.0.0-20230308081941-4dad7d3f5cb1 h1:zZ8LlzkS3ls1OBqQPvqpLSEyZjOrjNaO2ekBAXe7/uI=
+github.com/veezhang/nebula-importer/v4 v4.0.0-20230308081941-4dad7d3f5cb1/go.mod h1:6Z0GAlZdHWMQCyLkq377Li6ibIHkcwkzmBTEwIorAoU=
github.com/vesoft-inc/go-pkg v0.0.0-20230307111816-a43d6a342d23 h1:uiuSGlc1MTuEOy+VKBFxXA8WayxXeGigh6DT0On5r/I=
github.com/vesoft-inc/go-pkg v0.0.0-20230307111816-a43d6a342d23/go.mod h1:HCAXRhF2io+nPLQnl+RQ6XyVcp1Xdv6NgslXRBBCiEU=
github.com/vesoft-inc/nebula-go/v3 v3.4.0 h1:7q2DSW4QABwI2oGPSVuC+Ql7kGwj26G/YVPGD7gETys=
From c37f90c3862f71b5c3697fcab1fc0409bf07fbb7 Mon Sep 17 00:00:00 2001
From: hetao92 <18328704+hetao92@users.noreply.github.com>
Date: Thu, 9 Mar 2023 02:51:51 +0000
Subject: [PATCH 3/4] mod: add withhaeder & remove vidprefix
---
app/components/CSVPreviewLink/index.tsx | 1 -
app/config/locale/en-US.json | 1 -
app/config/locale/zh-CN.json | 1 -
.../FileList/PreviewFileModal/index.tsx | 6 +-
.../FileList/UploadConfigModal/index.tsx | 75 ++++++++++---------
app/pages/Import/FileList/index.tsx | 12 +--
.../SchemaConfig/FileMapping/index.tsx | 11 +--
app/stores/import.ts | 1 -
server/api/studio/internal/service/import.go | 1 -
.../internal/service/importer/taskmgr.go | 4 -
server/go.mod | 2 +-
server/go.sum | 4 +-
12 files changed, 51 insertions(+), 68 deletions(-)
diff --git a/app/components/CSVPreviewLink/index.tsx b/app/components/CSVPreviewLink/index.tsx
index 52999961..f3f6083d 100644
--- a/app/components/CSVPreviewLink/index.tsx
+++ b/app/components/CSVPreviewLink/index.tsx
@@ -21,7 +21,6 @@ const CSVPreviewLink = (props: IProps) => {
const [data, setData] = useState
([]);
const { intl } = useI18n();
const { readString } = usePapaParse();
-
useEffect(() => {
if(!file) return;
const { delimiter, sample } = file;
diff --git a/app/config/locale/en-US.json b/app/config/locale/en-US.json
index ab65fdc1..a8fb7149 100644
--- a/app/config/locale/en-US.json
+++ b/app/config/locale/en-US.json
@@ -217,7 +217,6 @@
"srcVidColumn": "Source VID column",
"dstVidColumn": "Destination VID column",
"vidFunction": "VID function",
- "vidPrefix": "VID prefix",
"concurrencyTip": "Number of NebulaGraph client concurrency.",
"batchSizeTip": "The number of statements inserting data in a batch.",
"retryTip": "Retry times of nGQL statement execution failures.",
diff --git a/app/config/locale/zh-CN.json b/app/config/locale/zh-CN.json
index 8ea46787..21991f22 100644
--- a/app/config/locale/zh-CN.json
+++ b/app/config/locale/zh-CN.json
@@ -217,7 +217,6 @@
"srcVidColumn": "起点 VID 列",
"dstVidColumn": "终点 VID 列",
"vidFunction": "VID 函数",
- "vidPrefix": "VID 前缀",
"concurrencyTip": "NebulaGraph 客户端并发数",
"batchSizeTip": "单批次插入数据的语句数量",
"retryTip": "nGQL 语句执行失败的重试次数",
diff --git a/app/pages/Import/FileList/PreviewFileModal/index.tsx b/app/pages/Import/FileList/PreviewFileModal/index.tsx
index 7312372b..12f72cab 100644
--- a/app/pages/Import/FileList/PreviewFileModal/index.tsx
+++ b/app/pages/Import/FileList/PreviewFileModal/index.tsx
@@ -1,5 +1,5 @@
import { useI18n } from '@vesoft-inc/i18n';
-import { Button, Input, Modal, Table, Form, Row, Col, message } from 'antd';
+import { Button, Input, Modal, Table, Form, Row, Col, message, Checkbox } from 'antd';
import { v4 as uuidv4 } from 'uuid';
import React, { PropsWithChildren, useCallback, useEffect } from 'react';
import { usePapaParse } from 'react-papaparse';
@@ -92,11 +92,11 @@ const PreviewFileModal = (props: IProps) => {
delimiter,
}}>
- {/*
+
{intl.get('import.hasHeader')}
- */}
+
diff --git a/app/pages/Import/FileList/UploadConfigModal/index.tsx b/app/pages/Import/FileList/UploadConfigModal/index.tsx
index 344128c2..571527bf 100644
--- a/app/pages/Import/FileList/UploadConfigModal/index.tsx
+++ b/app/pages/Import/FileList/UploadConfigModal/index.tsx
@@ -10,6 +10,7 @@ import { useStore } from '@app/stores';
import { observer, useLocalObservable } from 'mobx-react-lite';
import { ExclamationCircleFilled } from '@ant-design/icons';
import { observable } from 'mobx';
+import Checkbox, { CheckboxChangeEvent } from 'antd/lib/checkbox';
import styles from './index.module.less';
interface IProps {
visible: boolean;
@@ -38,8 +39,8 @@ const UploadConfigModal = (props: IProps) => {
data: [],
activeItem: null,
previewContent: [],
- // checkAll: false,
- // indeterminate: false,
+ checkAll: false,
+ indeterminate: false,
loading: false,
uploading: false,
setState: (obj) => Object.assign(state, obj),
@@ -73,36 +74,36 @@ const UploadConfigModal = (props: IProps) => {
});
}, []);
- // const onCheckAllChange = useCallback((e: CheckboxChangeEvent) => {
- // const { data, setState } = state;
- // const { checked } = e.target;
- // setState({
- // checkAll: checked,
- // indeterminate: false,
- // data: data.map(i => (i.withHeader = checked, i))
- // });
- // }, []);
+ const onCheckAllChange = useCallback((e: CheckboxChangeEvent) => {
+ const { data, setState } = state;
+ const { checked } = e.target;
+ setState({
+ checkAll: checked,
+ indeterminate: false,
+ data: data.map(i => (i.withHeader = checked, i))
+ });
+ }, []);
- // const updateItem = useCallback((e: CheckboxChangeEvent, item: StudioFile) => {
- // const { data, setState } = state;
- // const { checked } = e.target;
- // const nextData = data.map(i => (i === item && (i.withHeader = checked), i));
- // const checkedNum = data.reduce((n, file) => n + (file.withHeader & 1), 0);
- // setState({
- // checkAll: checkedNum === data.length,
- // indeterminate: !!checkedNum && checkedNum < data.length,
- // data: nextData
- // });
- // }, []);
+ const updateItem = useCallback((e: CheckboxChangeEvent, item: StudioFile) => {
+ const { data, setState } = state;
+ const { checked } = e.target;
+ const nextData = data.map(i => (i === item && (i.withHeader = checked), i));
+ const checkedNum = data.reduce((n, file) => n + (file.withHeader & 1), 0);
+ setState({
+ checkAll: checkedNum === data.length,
+ indeterminate: !!checkedNum && checkedNum < data.length,
+ data: nextData
+ });
+ }, []);
const deletePreviewFile = useCallback((e: React.MouseEvent, index: number) => {
const { activeItem, data, setState, previewContent } = state;
e.stopPropagation();
const isActive = activeItem?.uid === data[index].uid;
const newData = data.filter((_, i) => i !== index);
- // const checkedNum = data.reduce((n, file) => n + (file.withHeader & 1), 0);
+ const checkedNum = data.reduce((n, file) => n + (file.withHeader & 1), 0);
setState({
- // checkAll: checkedNum === newData.length && newData.length > 0,
- // indeterminate: !!checkedNum && checkedNum < newData.length,
+ checkAll: checkedNum === newData.length && newData.length > 0,
+ indeterminate: !!checkedNum && checkedNum < newData.length,
data: newData,
activeItem: isActive ? null : activeItem,
previewContent: isActive ? [] : previewContent,
@@ -170,7 +171,7 @@ const UploadConfigModal = (props: IProps) => {
if(!visible) {
return null;
}
- const { uploading, data, activeItem, previewContent, loading, setState } = state;
+ const { uploading, data, activeItem, previewContent, loading, setState, checkAll, indeterminate } = state;
const parseColumns = previewContent.length
? previewContent[0].map((header, index) => {
const textIndex = index;
@@ -186,18 +187,18 @@ const UploadConfigModal = (props: IProps) => {
{
title: intl.get('import.fileName'),
dataIndex: 'name',
- // align: 'center' as const,
- width: '60%'
+ align: 'center' as const,
+ width: '30%'
+ },
+ {
+ title: <>
+
+ {intl.get('import.withHeader')}
+ ,
+ key: 'withHeader',
+ width: '30%',
+ render: record => updateItem(e, record)}>{intl.get('import.hasHeader')},
},
- // {
- // title: <>
- //
- // {intl.get('import.withHeader')}
- // ,
- // key: 'withHeader',
- // width: '30%',
- // render: record => updateItem(e, record)}>{intl.get('import.hasHeader')},
- // },
{
title: <>
{intl.get('import.delimiter')}
diff --git a/app/pages/Import/FileList/index.tsx b/app/pages/Import/FileList/index.tsx
index 8541813b..12374e4d 100644
--- a/app/pages/Import/FileList/index.tsx
+++ b/app/pages/Import/FileList/index.tsx
@@ -26,13 +26,13 @@ const FileList = () => {
{
title: intl.get('import.fileName'),
dataIndex: 'name',
- width: '40%'
+ width: '50%'
+ },
+ {
+ title: intl.get('import.withHeader'),
+ dataIndex: 'withHeader',
+ render: value => value ? intl.get('import.hasHeader') : intl.get('import.noHeader'),
},
- // {
- // title: intl.get('import.withHeader'),
- // dataIndex: 'withHeader',
- // render: value => value ? intl.get('import.hasHeader') : intl.get('import.noHeader'),
- // },
{
title: intl.get('import.delimiter'),
dataIndex: 'delimiter',
diff --git a/app/pages/Import/TaskCreate/SchemaConfig/FileMapping/index.tsx b/app/pages/Import/TaskCreate/SchemaConfig/FileMapping/index.tsx
index 46114553..c5bae406 100644
--- a/app/pages/Import/TaskCreate/SchemaConfig/FileMapping/index.tsx
+++ b/app/pages/Import/TaskCreate/SchemaConfig/FileMapping/index.tsx
@@ -1,4 +1,4 @@
-import { Collapse, Input, Select, Table, Tooltip } from 'antd';
+import { Collapse, Select, Table, Tooltip } from 'antd';
import React, { useMemo } from 'react';
import { observer } from 'mobx-react-lite';
import cls from 'classnames';
@@ -53,10 +53,6 @@ const VIDSetting = observer((props: {
{intl.get('import.vidFunction')}
{intl.get('import.vidFunctionTip')}