disttask

Documentation

Index

• Constants
• Variables
• func GetK8sInstanceKey(queueName string) *config.InstanceType
• func NewDisttaskEngine(conf EngineConfig, crmMgr, k8sCrmMgr, dcMacMgr crm.HandlerWithUser, ...) (engine.Engine, error)
• type CCacheStats
  • func (cs CCacheStats) Dump() []byte
• type CombinedProject
• type CustomData
• type DeltaInfoStats
• type EngineConfig
• type ExtraData
• type K8sClusterInfo
• type Message
• type MessageRecordStats
  • func (mrd MessageRecordStats) Dump() []byte
• type MessageTaskStats
• type MessageType
• type MySQL
  • func NewMySQL(conf engine.MySQLConf) (MySQL, error)
• type SummaryResult
• type SummaryResultByUser
• type TableProjectInfo
  • func (tpi TableProjectInfo) TableName() string
• type TableProjectSetting
  • func (tps TableProjectSetting) TableName() string
• type TableTask
  • func (tt TableTask) TableName() string
• type TableWhitelist
  • func (twl TableWhitelist) TableName() string
• type TableWorkStats
  • func (tts *TableWorkStats) TableName() string
• type TableWorker
  • func (tg *TableWorker) CheckData() error
  • func (tg TableWorker) TableName() string

Constants

const (
	// EngineName define the engine name
	EngineName = "disttask"
)

Variables

var (
	EmptyJobs = compress.ToBase64String([]byte("[]"))
)

EmptyJobs define

Functions

func GetK8sInstanceKey

func GetK8sInstanceKey(queueName string) *config.InstanceType

GetK8sInstanceKey get instance type from queueName

func NewDisttaskEngine

func NewDisttaskEngine(
	conf EngineConfig,
	crmMgr, k8sCrmMgr, dcMacMgr crm.HandlerWithUser,
	k8sListCrmMgr map[string]crm.HandlerWithUser,
	directMgr direct.HandleWithUser) (engine.Engine, error)

NewDisttaskEngine get a new distTask engine

Types

type CCacheStats

type CCacheStats struct {
	CacheDir                  string `json:"cache_dir"`
	PrimaryConfig             string `json:"primary_config"`
	SecondaryConfig           string `json:"secondary_config"`
	DirectHit                 int    `json:"cache_direct_hit"`
	PreprocessedHit           int    `json:"cache_preprocessed_hit"`
	CacheMiss                 int    `json:"cache_miss"`
	CalledForLink             int    `json:"called_for_link"`
	CalledForPreProcessing    int    `json:"called_for_processing"`
	UnsupportedSourceLanguage int    `json:"unsupported_source_language"`
	NoInputFile               int    `json:"no_input_file"`
	FilesInCache              int    `json:"files_in_cache"`
	CacheSize                 string `json:"cache_size"`
	MaxCacheSize              string `json:"max_cache_size"`
}

CCacheStats describe the ccache stats data from 'ccache -s'.

func (CCacheStats) Dump

func (cs CCacheStats) Dump() []byte

dump the struct data into byte

type CombinedProject

type CombinedProject struct {
	*TableProjectSetting
	*TableProjectInfo
}

CombinedProject generate project_settings and project_records

type CustomData

type CustomData struct {
	WorkerVersion     string            `json:"worker_version"`
	Environments      map[string]string `json:"environments"`
	UnsetEnvironments []string          `json:"unset_environments"`
	JobServer         int               `json:"job_server"`
	ExtraWorkerData   string            `json:"extra_worker_data"`
	ExtraProjectData  string            `json:"extra_project_data"`
	SupportAbsPath    bool              `json:"support_abs_path"`
}

CustomData describe the detail data of dist task.

type DeltaInfoStats

type DeltaInfoStats struct {
	CompileFilesOK      int64
	CompileFilesErr     int64
	CompileFilesTimeout int64
	CompileUnits        float64
}

DeltaInfoStats describe the project info delta data.

type EngineConfig

type EngineConfig struct {
	engine.MySQLConf
	Rd                     rd.RegisterDiscover
	QueueResourceAllocater map[string]config.ResourceAllocater

	JobServerTimesToCPU float64
	QueueShareType      map[string]engine.QueueShareType

	// mesos cluster info
	CRMClusterID      string
	CRMCPUPerInstance float64
	CRMMemPerInstance float64

	// k8s cluster info
	K8SCRMClusterID      string
	K8SCRMCPUPerInstance float64
	K8SCRMMemPerInstance float64

	// k8s cluster list info
	K8SClusterList map[string]K8sClusterInfo

	// dc_mac cluster info
	VMCRMClusterID      string
	VMCRMCPUPerInstance float64
	VMCRMMemPerInstance float64

	Brokers []config.EngineDisttaskBrokerConfig
}

EngineConfig define engine config

type ExtraData

type ExtraData struct {
	User           string `json:"user"`
	RunDir         string `json:"run_dir"`
	Params         string `json:"params"` //自定义参数
	Cmd            string `json:"cmd"`
	ProcessPerUnit int    `json:"process_per_unit"`

	// command define the target to be called, such as make, bazel, /data/custom/make etc.
	BoosterType string `json:"BoosterType,omitempty"`

	// extra_vars includes the extra params need by client
	ExtraVars taskClientExtra `json:"extra_vars,omitempty"`
}

ExtraData describe the data in task creation from client.

type K8sClusterInfo

type K8sClusterInfo struct {
	K8SCRMClusterID      string
	K8SCRMCPUPerInstance float64
	K8SCRMMemPerInstance float64
}

K8sClusterInfo define

type Message

type Message struct {
	Type MessageType `json:"type"`

	// typo "task_tats", should be "task_stats", but for old version protocol usage, just keep it.
	MessageTaskStats   MessageTaskStats   `json:"task_tats"`
	MessageRecordStats MessageRecordStats `json:"ccache_stats"`
}

Message describe the data format from SendMessage caller.

type MessageRecordStats

type MessageRecordStats struct {
	Message     string      `json:"message"`
	CCacheStats CCacheStats `json:"ccache_stats"`
}

MessageRecordStats describe the message data of type record stats.

func (MessageRecordStats) Dump

func (mrd MessageRecordStats) Dump() []byte

dump the struct data into byte

type MessageTaskStats

type MessageTaskStats struct {
	WorkID           string `json:"work_id"`
	TaskID           string `json:"task_id"`
	Scene            string `json:"scene"`
	Success          bool   `json:"success"`
	JobRemoteOK      int    `json:"job_remote_ok"`
	JobRemoteError   int    `json:"job_remote_error"`
	JobLocalOK       int    `json:"job_local_ok"`
	JobLocalError    int    `json:"job_local_error"`
	StartTime        int64  `json:"start_time"`
	EndTime          int64  `json:"end_time"`
	RegisteredTime   int64  `json:"registered_time"`
	UnregisteredTime int64  `json:"unregistered_time"`
	Jobs             string `json:"jobs"`
}

MessageTaskStats describe the message of uploading task stats

type MessageType

type MessageType int

MessageType define

const (
	// MessageTypeTaskStats means this message is about record task stats from client.
	MessageTypeTaskStats MessageType = iota
	//MessageTypeRecordStats means
	MessageTypeRecordStats
)

type MySQL

type MySQL interface {
	// get db operator
	GetDB() *gorm.DB

	ListTask(opts commonMySQL.ListOptions) ([]*TableTask, int64, error)
	GetTask(taskID string) (*TableTask, error)
	PutTask(task *TableTask) error
	UpdateTask(taskID string, task map[string]interface{}) error
	DeleteTask(taskID string) error

	ListProject(opts commonMySQL.ListOptions) ([]*CombinedProject, int64, error)

	ListProjectInfo(opts commonMySQL.ListOptions) ([]*TableProjectInfo, int64, error)
	GetProjectInfo(projectID string) (*TableProjectInfo, error)
	PutProjectInfo(projectInfo *TableProjectInfo) error
	UpdateProjectInfo(projectID string, projectInfo map[string]interface{}) error
	DeleteProjectInfo(projectID string) error
	AddProjectInfoStats(projectID string, delta DeltaInfoStats) error

	ListProjectSetting(opts commonMySQL.ListOptions) ([]*TableProjectSetting, int64, error)
	GetProjectSetting(projectID string) (*TableProjectSetting, error)
	PutProjectSetting(projectSetting *TableProjectSetting) error
	UpdateProjectSetting(projectID string, projectSetting map[string]interface{}) error
	DeleteProjectSetting(projectID string) error
	CreateOrUpdateProjectSetting(projectSetting *TableProjectSetting, projectSettingRaw map[string]interface{}) error

	ListWhitelist(opts commonMySQL.ListOptions) ([]*TableWhitelist, int64, error)
	GetWhitelist(key engine.WhiteListKey) (*TableWhitelist, error)
	PutWhitelist(wll []*TableWhitelist) error
	UpdateWhitelist(key engine.WhiteListKey, wll []map[string]interface{}) error
	DeleteWhitelist(keys []*engine.WhiteListKey) error

	ListWorker(opts commonMySQL.ListOptions) ([]*TableWorker, int64, error)
	GetWorker(version, scene string) (*TableWorker, error)
	PutWorker(worker *TableWorker) error
	UpdateWorker(version, scene string, worker map[string]interface{}) error
	DeleteWorker(version, scene string) error

	ListWorkStats(opts commonMySQL.ListOptions) ([]*TableWorkStats, int64, error)
	// GetWorkStats(id int) (*TableWorkStats, error)
	GetWorkStats(taskID, workID string) (*TableWorkStats, error)
	PutWorkStats(stats *TableWorkStats) error
	UpdateWorkStats(id int, stats map[string]interface{}) error
	DeleteWorkStats(id int) error

	SummaryTaskRecords(opts commonMySQL.ListOptions) ([]*SummaryResult, int64, error)
	SummaryTaskRecordsByUser(opts commonMySQL.ListOptions) ([]*SummaryResultByUser, int64, error)
}

MySQL describe the full operations to mysql databases need by engine.

func NewMySQL

func NewMySQL(conf engine.MySQLConf) (MySQL, error)

NewMySQL get new mysql instance with connected orm operator.

type SummaryResult

type SummaryResult struct {
	Day               string  `json:"day"`
	ProjectID         string  `json:"project_id"`
	TotalTime         float64 `json:"total_time"`
	TotalTimeWithCPU  float64 `json:"total_time_with_cpu"`
	TotalRecordNumber int     `json:"total_record_number"`
}

SummaryResult generate summary data

type SummaryResultByUser

type SummaryResultByUser struct {
	Day               string  `json:"day"`
	ProjectID         string  `json:"project_id"`
	User              string  `json:"user"`
	TotalTime         float64 `json:"total_time"`
	TotalTimeWithCPU  float64 `json:"total_time_with_cpu"`
	TotalRecordNumber int     `json:"total_record_number"`
}

SummaryResultByUser generate summary data group by user

type TableProjectInfo

type TableProjectInfo struct {
	engine.TableProjectInfoBasic

	CompileFilesOK      int64   `gorm:"column:compile_files_ok" json:"compile_files_ok"`
	CompileFilesErr     int64   `gorm:"column:compile_files_err" json:"compile_files_err"`
	CompileFilesTimeout int64   `gorm:"column:compile_files_timeout" json:"compile_files_timeout"`
	CompileUnits        float64 `gorm:"column:compile_units" json:"compile_units"`
}

TableProjectInfo describe the db columns of project info. It will inherit the ProjectInfoBasic.

func (TableProjectInfo) TableName

func (tpi TableProjectInfo) TableName() string

TableName specific table name.

type TableProjectSetting

type TableProjectSetting struct {
	engine.TableProjectBasic

	RequestCPU float64 `gorm:"column:request_cpu" json:"request_cpu"`
	LeastCPU   float64 `gorm:"column:least_cpu" json:"least_cpu"`
	SuggestCPU float64 `gorm:"column:suggest_cpu;default:0" json:"suggest_cpu"`

	WorkerVersion string `gorm:"column:worker_version" json:"worker_version"`
	Scene         string `gorm:"column:scene" json:"scene"`
	BanAllBooster bool   `gorm:"column:ban_all_booster;default:false" json:"ban_all_booster"`

	// reserved for different project
	Extra string `gorm:"column:extra" json:"extra"`
}

TableProjectSetting describe the db columns of project setting. It will inherit the ProjectBasic.

func (TableProjectSetting) TableName

func (tps TableProjectSetting) TableName() string

TableName specific table name.

type TableTask

type TableTask struct {
	engine.TableTaskBasic

	// client
	SourceIP           string `gorm:"column:source_ip" json:"source_ip"`
	SourceCPU          int    `gorm:"column:source_cpu" json:"source_cpu"`
	User               string `gorm:"column:user" json:"user"`
	Params             string `gorm:"column:params" sql:"type:text" json:"params"`
	Cmd                string `gorm:"column:cmd" sql:"type:text" json:"cmd"`
	Env                string `gorm:"column:env" sql:"type:text" json:"env"`
	RunDir             string `gorm:"column:run_dir" sql:"type:text" json:"run_dir"`
	BoosterType        string `gorm:"column:booster_type" json:"booster_type"`
	BanAllBooster      bool   `gorm:"column:ban_all_booster;default:false" json:"ban_all_booster"`
	ExtraClientSetting string `gorm:"column:extra_client_setting" sql:"type:text" json:"extra_client_setting"`

	// workers
	Workers string `gorm:"column:workers" sql:"type:text" json:"workers"`

	// stats
	WorkerCount int     `gorm:"column:worker_count" json:"worker_count"`
	CPUTotal    float64 `gorm:"column:cpu_total" json:"cpu_total"`
	MemTotal    float64 `gorm:"column:mem_total" json:"mem_total"`
	SucceedNum  int64   `gorm:"column:succeed_num" json:"succeed_num"`
	FailedNum   int64   `gorm:"column:failed_num" json:"failed_num"`
	StatDetail  string  `gorm:"column:stat_detail" json:"stat_detail"`
	ExtraRecord string  `gorm:"column:extra_record" sql:"type:text" json:"extra_record"`

	// resource manager
	ClusterID             string  `gorm:"column:cluster_id" json:"cluster_id"`
	AppName               string  `gorm:"column:app_name" json:"app_name"`
	Namespace             string  `gorm:"column:namespace" json:"namespace"`
	Image                 string  `gorm:"column:image" sql:"type:text" json:"image"`
	Instance              int     `gorm:"column:instance" json:"instance"`
	RequestInstance       int     `gorm:"column:request_instance" json:"request_instance"`
	LeastInstance         int     `gorm:"column:least_instance" json:"least_instance"`
	RequestCPUPerUnit     float64 `gorm:"column:request_cpu_per_unit" json:"request_cpu_per_unit"`
	RequestMemPerUnit     float64 `gorm:"column:request_mem_per_unit" json:"request_mem_per_unit"`
	RequestProcessPerUnit int     `gorm:"column:request_process_per_unit" json:"request_process_per_unit"`

	// inherit from project setting or worker setting
	RequestCPU          float64 `gorm:"column:request_cpu" json:"request_cpu"`
	LeastCPU            float64 `gorm:"column:least_cpu" json:"least_cpu"`
	WorkerVersion       string  `gorm:"column:worker_version" json:"worker_version"`
	Scene               string  `gorm:"column:scene" json:"scene"`
	ExtraProjectSetting string  `gorm:"column:extra_project_setting" sql:"type:text" json:"extra_project_setting"`
	ExtraWorkerSetting  string  `gorm:"column:extra_worker_setting" sql:"type:text" json:"extra_worker_setting"`
}

TableTask describe the db columns of Task. It will inherit the TaskBasic.

func (TableTask) TableName

func (tt TableTask) TableName() string

TableName specific table name.

type TableWhitelist

type TableWhitelist struct {
	engine.TableWhitelistBasic
}

TableWhitelist describe the db columns of whitelist. It will inherit the WhitelistBasic.

func (TableWhitelist) TableName

func (twl TableWhitelist) TableName() string

TableName specific table name.

type TableWorkStats

type TableWorkStats struct {
	engine.TableBasic

	ID               int    `gorm:"column:id;AUTO_INCREMENT" json:"id"`
	ProjectID        string `gorm:"column:project_id" json:"project_id"`
	TaskID           string `gorm:"column:task_id;index" json:"task_id"`
	WorkID           string `gorm:"column:work_id;index" json:"work_id"`
	Scene            string `gorm:"column:scene;index" json:"scene"`
	Success          bool   `gorm:"column:success;index" json:"success"`
	JobRemoteOK      int    `gorm:"column:job_remote_ok" json:"job_remote_ok"`
	JobRemoteError   int    `gorm:"column:job_remote_error" json:"job_remote_error"`
	JobLocalOK       int    `gorm:"column:job_local_ok" json:"job_local_ok"`
	JobLocalError    int    `gorm:"column:job_local_error" json:"job_local_error"`
	StartTime        int64  `gorm:"column:start_time" json:"start_time"`
	EndTime          int64  `gorm:"column:end_time" json:"end_time"`
	RegisteredTime   int64  `gorm:"column:registered_time" json:"registered_time"`
	UnregisteredTime int64  `gorm:"column:unregistered_time" json:"unregistered_time"`
	JobStats         string `gorm:"column:job_stats;type:longtext" json:"job_stats"`
}

TableWorkStats

func (*TableWorkStats) TableName

func (tts *TableWorkStats) TableName() string

TableName specific table name.

type TableWorker

type TableWorker struct {
	engine.TableBasic

	WorkerVersion string `gorm:"column:worker_version;primary_key" json:"worker_version"`
	Scene         string `gorm:"column:scene;primary_key" json:"scene"`
	Image         string `gorm:"column:image" sql:"type:text" json:"image"`

	// reserved
	Extra string `gorm:"column:extra" json:"extra"`
}

TableWorker describe the db columns of worker. It inherit directly from Basic.

func (*TableWorker) CheckData

func (tg *TableWorker) CheckData() error

CheckData check if worker data valid.

func (TableWorker) TableName

func (tg TableWorker) TableName() string

TableName specific table name.