importer

Documentation

Index

• func CreateConfigFile(taskDir string, cfgBytes []byte) (string, error)
• func CreateNewTaskDir(rootDir string, id string) (string, error)
• func DeleteImportTask(tasksDir, taskID, address, username string) error
• func GetImportTask(taskID, address, username string) (*types.GetImportTaskData, error)
• func GetManyImportTask(address, username, space string, pageIndex, pageSize int, status string) (*types.GetManyImportTaskData, error)
• func InitTask()
• func NewTaskMgr(serverName string, tackConfig *types.TaskConfig)
• func StartImport(taskID string) (err error)
• func StopImportTask(taskID, address, username string) error
• type Client
• type ImportResult
• type Task
  • func UnmarshalTask(byts []byte) (t *Task, err error)
  • func (t *Task) Marshal() (byts []byte, err error)
  • func (t *Task) ToImportTaskData() (result *types.GetImportTaskData, err error)
  • func (t *Task) UpdateQueryStats() error
• type TaskMgr
  • func GetTaskMgr() *TaskMgr
  • func (mgr *TaskMgr) AbortTask(taskID, msg string) (err error)
  • func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error
  • func (mgr *TaskMgr) FinishTask(taskID, status string) (err error)
  • func (mgr *TaskMgr) GetDcron() *dcron.Dcron
  • func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool)
  • func (mgr *TaskMgr) GetTaskInfoByName(taskName string) (task *db.TaskInfo, err error)
  • func (mgr *TaskMgr) NewTask(id, cron, host, user, taskName string, rawConfig string, ...) (*Task, error)
  • func (mgr *TaskMgr) NewTaskDraft(id, host, user, taskName, space, rawCfg string) error
  • func (mgr *TaskMgr) NewTaskEffect(taskEffect *db.TaskEffect) error
  • func (mgr *TaskMgr) PutTask(taskID string, task *Task)
  • func (mgr *TaskMgr) StartTask(info *db.TaskInfo) error
  • func (mgr *TaskMgr) Stop()
  • func (mgr *TaskMgr) StopTask(taskID string) error
  • func (mgr *TaskMgr) StorePartTaskLog(taskID string) error
  • func (mgr *TaskMgr) TurnDraftToTask(id, cron, taskName string, rawCfg string, cfg *types.ImportTaskV2Config) (*Task, error)
  • func (mgr *TaskMgr) UpdateTask(task *Task) error
  • func (mgr *TaskMgr) UpdateTaskDraft(id, taskName, space, rawCfg string) error
  • func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error
  • func (mgr *TaskMgr) UpdateTaskInfoById(task *Task) error

Functions

func CreateConfigFile

func CreateConfigFile(taskDir string, cfgBytes []byte) (string, error)

func CreateNewTaskDir

func CreateNewTaskDir(rootDir string, id string) (string, error)

func DeleteImportTask

func DeleteImportTask(tasksDir, taskID, address, username string) error

func GetImportTask

func GetImportTask(taskID, address, username string) (*types.GetImportTaskData, error)

func GetManyImportTask

func GetManyImportTask(
	address, username, space string,
	pageIndex, pageSize int,
	status string) (*types.GetManyImportTaskData, error)

func InitTask

func InitTask()

func NewTaskMgr

func NewTaskMgr(serverName string, tackConfig *types.TaskConfig)

func StartImport

func StartImport(taskID string) (err error)

func StopImportTask

func StopImportTask(taskID, address, username string) error

Types

type Client

type Client struct {
	Cfg        *types.ImportTaskV2Config `json:"cfg,omitempty"`
	Logger     logger.Logger             `json:"logger,omitempty"`
	Manager    manager.Manager           `json:"manager,omitempty"`
	HasStarted bool                      `json:"has_started,omitempty"`
}

type ImportResult

type ImportResult struct {
	TaskId      string `json:"taskId"`
	TimeCost    string `json:"timeCost"` // Milliseconds
	FailedRows  int64  `json:"failedRows"`
	ErrorResult struct {
		ErrorCode int    `json:"errorCode"`
		ErrorMsg  string `json:"errorMsg"`
	}
}

type Task

type Task struct {
	Client   *Client      `json:"client,omitempty"`
	TaskInfo *db.TaskInfo `json:"task_info,omitempty"`
}

func UnmarshalTask

func UnmarshalTask(byts []byte) (t *Task, err error)

func (*Task) Marshal

func (t *Task) Marshal() (byts []byte, err error)

func (*Task) ToImportTaskData

func (t *Task) ToImportTaskData() (result *types.GetImportTaskData, err error)

func (*Task) UpdateQueryStats

func (t *Task) UpdateQueryStats() error

type TaskMgr

type TaskMgr struct {
	// contains filtered or unexported fields
}

func GetTaskMgr

func GetTaskMgr() *TaskMgr

func (*TaskMgr) AbortTask

func (mgr *TaskMgr) AbortTask(taskID, msg string) (err error)

func (*TaskMgr) DelTask

func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error

func (*TaskMgr) FinishTask

func (mgr *TaskMgr) FinishTask(taskID, status string) (err error)

FinishTask will query task stats

• delete task in the map
• update taskInfo in db
• update taskEffect in db

func (*TaskMgr) GetDcron

func (mgr *TaskMgr) GetDcron() *dcron.Dcron

func (*TaskMgr) GetTask

func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool)

GetTask get task from map and local sql

func (*TaskMgr) GetTaskInfoByName

func (mgr *TaskMgr) GetTaskInfoByName(taskName string) (task *db.TaskInfo, err error)

func (*TaskMgr) NewTask

func (mgr *TaskMgr) NewTask(id, cron, host, user, taskName string, rawConfig string, cfg *types.ImportTaskV2Config) (*Task, error)

func (*TaskMgr) NewTaskDraft

func (mgr *TaskMgr) NewTaskDraft(id, host, user, taskName, space, rawCfg string) error

func (*TaskMgr) NewTaskEffect

func (mgr *TaskMgr) NewTaskEffect(taskEffect *db.TaskEffect) error

func (*TaskMgr) PutTask

func (mgr *TaskMgr) PutTask(taskID string, task *Task)

PutTask put task into cache map

func (*TaskMgr) StartTask

func (mgr *TaskMgr) StartTask(info *db.TaskInfo) error

func (*TaskMgr) Stop

func (mgr *TaskMgr) Stop()

func (*TaskMgr) StopTask

func (mgr *TaskMgr) StopTask(taskID string) error

StopTask will change the task status to `Stopped`, and then call FinishTask

func (*TaskMgr) StorePartTaskLog

func (mgr *TaskMgr) StorePartTaskLog(taskID string) error

func (*TaskMgr) TurnDraftToTask

func (mgr *TaskMgr) TurnDraftToTask(id, cron, taskName string, rawCfg string, cfg *types.ImportTaskV2Config) (*Task, error)

func (*TaskMgr) UpdateTask

func (mgr *TaskMgr) UpdateTask(task *Task) error

func (*TaskMgr) UpdateTaskDraft

func (mgr *TaskMgr) UpdateTaskDraft(id, taskName, space, rawCfg string) error

func (*TaskMgr) UpdateTaskInfo

func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error

UpdateTaskInfo will query task stats, update task in the map and update the taskInfo in local sql

func (*TaskMgr) UpdateTaskInfoById

func (mgr *TaskMgr) UpdateTaskInfoById(task *Task) error

UpdateTaskInfoById Update task info by task id