From 602c8f611f8bcdfc93b395363d2402128c919fb4 Mon Sep 17 00:00:00 2001 From: songjc <969378911@qq.com> Date: Wed, 6 Sep 2023 15:51:11 +0800 Subject: [PATCH] advisor init --- advisor/README.md | 2 + advisor/internal/config/config.go | 23 +++++ advisor/internal/globals/globals.go | 9 ++ advisor/internal/reporter/reporter.go | 84 +++++++++++++++++++ advisor/internal/services/advisor.go | 12 +++ advisor/internal/services/service.go | 15 ++++ advisor/internal/task/schedule_scheme.go | 73 ++++++++++++++++ advisor/internal/task/task.go | 28 +++++++ advisor/main.go | 74 ++++++++++++++++ common/pkgs/mq/advisor/apis.go | 39 +++++++++ common/pkgs/mq/advisor/client.go | 59 +++++++++++++ common/pkgs/mq/advisor/server.go | 70 ++++++++++++++++ common/pkgs/mq/advisor/task/task.go | 73 ++++++++++++++++ common/pkgs/mq/manager/api.go | 53 ++++++++++-- common/pkgs/mq/manager/server.go | 3 +- executor/README.md | 2 +- executor/internal/reporter/reporter.go | 2 +- ...scheduler_task.go => pcm_schedule_task.go} | 0 18 files changed, 609 insertions(+), 12 deletions(-) create mode 100644 advisor/README.md create mode 100644 advisor/internal/config/config.go create mode 100644 advisor/internal/globals/globals.go create mode 100644 advisor/internal/reporter/reporter.go create mode 100644 advisor/internal/services/advisor.go create mode 100644 advisor/internal/services/service.go create mode 100644 advisor/internal/task/schedule_scheme.go create mode 100644 advisor/internal/task/task.go create mode 100644 advisor/main.go create mode 100644 common/pkgs/mq/advisor/apis.go create mode 100644 common/pkgs/mq/advisor/client.go create mode 100644 common/pkgs/mq/advisor/server.go create mode 100644 common/pkgs/mq/advisor/task/task.go rename executor/internal/task/{pcm_scheduler_task.go => pcm_schedule_task.go} (100%) diff --git a/advisor/README.md b/advisor/README.md new file mode 100644 index 0000000..161d041 --- /dev/null +++ b/advisor/README.md @@ -0,0 +1,2 @@ +# scheduler-advisor + diff --git a/advisor/internal/config/config.go b/advisor/internal/config/config.go new file mode 100644 index 0000000..6214244 --- /dev/null +++ b/advisor/internal/config/config.go @@ -0,0 +1,23 @@ +package config + +import ( + log "gitlink.org.cn/cloudream/common/pkgs/logger" + c "gitlink.org.cn/cloudream/common/utils/config" + mymq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq" +) + +type Config struct { + Logger log.Config `json:"logger"` + ReportIntervalSec int `json:"reportIntervalSec"` + RabbitMQ mymq.Config `json:"rabbitMQ"` +} + +var cfg Config + +func Init() error { + return c.DefaultLoad("advisor", &cfg) +} + +func Cfg() *Config { + return &cfg +} diff --git a/advisor/internal/globals/globals.go b/advisor/internal/globals/globals.go new file mode 100644 index 0000000..c7be90f --- /dev/null +++ b/advisor/internal/globals/globals.go @@ -0,0 +1,9 @@ +package globals + +import "github.com/google/uuid" + +var AdvisorID string + +func Init() { + AdvisorID = uuid.NewString() +} diff --git a/advisor/internal/reporter/reporter.go b/advisor/internal/reporter/reporter.go new file mode 100644 index 0000000..b84f071 --- /dev/null +++ b/advisor/internal/reporter/reporter.go @@ -0,0 +1,84 @@ +package reporter + +import ( + "fmt" + "sync" + "time" + + "gitlink.org.cn/cloudream/common/pkgs/logger" + "gitlink.org.cn/cloudream/scheduler/common/globals" + advtsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/advisor/task" + mgrmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/manager" +) + +type Reporter struct { + advisorID string + reportInterval time.Duration + taskStatus map[string]advtsk.TaskStatus + taskStatusLock sync.Mutex + reportNow chan bool +} + +func NewReporter(advisorID string, reportInterval time.Duration) Reporter { + return Reporter{ + advisorID: advisorID, + reportInterval: reportInterval, + taskStatus: make(map[string]advtsk.TaskStatus), + reportNow: make(chan bool), + } +} + +func (r *Reporter) Report(taskID string, taskStatus advtsk.TaskStatus) { + r.taskStatusLock.Lock() + defer r.taskStatusLock.Unlock() + + r.taskStatus[taskID] = taskStatus +} + +func (r *Reporter) ReportNow() { + select { + case r.reportNow <- true: + default: + } +} + +func (r *Reporter) Serve() error { + magCli, err := globals.ManagerMQPool.Acquire() + if err != nil { + return fmt.Errorf("new manager client: %w", err) + } + defer magCli.Close() + + ticker := time.NewTicker(r.reportInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + case <-r.reportNow: + ticker.Reset(r.reportInterval) + } + + r.taskStatusLock.Lock() + var taskStatus []mgrmq.AdvisorTaskStatus + for taskID, status := range r.taskStatus { + taskStatus = append(taskStatus, mgrmq.NewAdvisorTaskStatus(taskID, status)) + } + r.taskStatus = make(map[string]advtsk.TaskStatus) + r.taskStatusLock.Unlock() + + _, err := magCli.ReportAdvisorTaskStatus(mgrmq.NewReportAdvisorTaskStatus(r.advisorID, taskStatus)) + if err != nil { + logger.Warnf("reporting to manager: %s", err.Error()) + + //若上报失败，数据应保留 + r.taskStatusLock.Lock() + for _, ts := range taskStatus { + if _, exists := r.taskStatus[ts.TaskID]; !exists { + r.taskStatus[ts.TaskID] = ts.Status + } + } + r.taskStatusLock.Unlock() + } + } +} diff --git a/advisor/internal/services/advisor.go b/advisor/internal/services/advisor.go new file mode 100644 index 0000000..056769b --- /dev/null +++ b/advisor/internal/services/advisor.go @@ -0,0 +1,12 @@ +package services + +import ( + "gitlink.org.cn/cloudream/common/pkgs/mq" + schtsk "gitlink.org.cn/cloudream/scheduler/advisor/internal/task" + advmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/advisor" +) + +func (svc *Service) StartGetScheduleScheme(msg *advmq.StartGetScheduleScheme) (*advmq.StartGetScheduleSchemeResp, *mq.CodeMessage) { + tsk := svc.taskManager.StartNew(schtsk.NewGetScheduleScheme()) + return mq.ReplyOK(advmq.NewStartGetScheduleSchemeResp(tsk.ID())) +} diff --git a/advisor/internal/services/service.go b/advisor/internal/services/service.go new file mode 100644 index 0000000..6ce06d6 --- /dev/null +++ b/advisor/internal/services/service.go @@ -0,0 +1,15 @@ +package services + +import ( + "gitlink.org.cn/cloudream/scheduler/advisor/internal/task" +) + +type Service struct { + taskManager *task.Manager +} + +func NewService(taskMgr *task.Manager) *Service { + return &Service{ + taskManager: taskMgr, + } +} diff --git a/advisor/internal/task/schedule_scheme.go b/advisor/internal/task/schedule_scheme.go new file mode 100644 index 0000000..ce5a105 --- /dev/null +++ b/advisor/internal/task/schedule_scheme.go @@ -0,0 +1,73 @@ +package task + +import ( + "time" + + "gitlink.org.cn/cloudream/common/pkgs/logger" + "gitlink.org.cn/cloudream/common/pkgs/task" + exectsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor/task" +) + +type GetScheduleScheme struct { +} + +func NewGetScheduleScheme() *GetScheduleScheme { + return &GetScheduleScheme{} +} + +func (t *GetScheduleScheme) Execute(task *task.Task[TaskContext], ctx TaskContext, complete CompleteFn) { + log := logger.WithType[GetScheduleScheme]("Task") + log.Debugf("begin") + defer log.Debugf("end") + + err := t.do(task.ID(), ctx) + if err != nil { + //TODO 若任务失败，上报的状态failed字段根据情况修改 + ctx.reporter.Report(task.ID(), exectsk.NewScheduleTaskStatus("failed", err.Error(), 0)) + } + ctx.reporter.ReportNow() + + complete(err, CompleteOption{ + RemovingDelay: time.Minute, + }) +} + +func (t *GetScheduleScheme) do(taskID string, ctx TaskContext) error { + // pcmCli, err := globals.PCMPool.Acquire() + // if err != nil { + // return fmt.Errorf("new pcm client: %w", err) + // } + // defer pcmCli.Close() + + // resp, err := pcmCli.ScheduleTask(pcm.ScheduleTaskReq{ + + // }) + + // if err != nil { + // return err + // } + + // var prevStatus string + // for { + // tsResp, err := pcmCli.GetTaskStatus(pcm.GetTaskStatusReq{ + // NodeID: t.nodeID, + // PCMJobID: resp.PCMJobID, + // }) + // if err != nil { + // return err + // } + + // if tsResp.Status != prevStatus { + // ctx.reporter.Report(taskID, exectsk.NewScheduleTaskStatus(tsResp.Status, "", resp.PCMJobID)) + // } + + // prevStatus = tsResp.Status + + // // TODO 根据接口result返回情况修改 + // // 根据返回的result判定任务是否完成，若完成 跳出循环，结束任务 + // if tsResp.Status == "Completed" { + // return nil + // } + // } + return nil +} diff --git a/advisor/internal/task/task.go b/advisor/internal/task/task.go new file mode 100644 index 0000000..fb70a9d --- /dev/null +++ b/advisor/internal/task/task.go @@ -0,0 +1,28 @@ +package task + +import ( + "gitlink.org.cn/cloudream/common/pkgs/task" + reporter "gitlink.org.cn/cloudream/scheduler/advisor/internal/reporter" +) + +type TaskContext struct { + reporter *reporter.Reporter +} + +// 需要在Task结束后主动调用，completing函数将在Manager加锁期间被调用， +// 因此适合进行执行结果的设置 +type CompleteFn = task.CompleteFn + +type Manager = task.Manager[TaskContext] + +type TaskBody = task.TaskBody[TaskContext] + +type Task = task.Task[TaskContext] + +type CompleteOption = task.CompleteOption + +func NewManager(reporter *reporter.Reporter) Manager { + return task.NewManager(TaskContext{ + reporter: reporter, + }) +} diff --git a/advisor/main.go b/advisor/main.go new file mode 100644 index 0000000..2c7e923 --- /dev/null +++ b/advisor/main.go @@ -0,0 +1,74 @@ +package main + +import ( + "fmt" + "os" + "time" + + "gitlink.org.cn/cloudream/common/pkgs/logger" + "gitlink.org.cn/cloudream/scheduler/advisor/internal/config" + myglbs "gitlink.org.cn/cloudream/scheduler/advisor/internal/globals" + "gitlink.org.cn/cloudream/scheduler/advisor/internal/reporter" + "gitlink.org.cn/cloudream/scheduler/advisor/internal/services" + "gitlink.org.cn/cloudream/scheduler/advisor/internal/task" + execmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor" +) + +func main() { + err := config.Init() + if err != nil { + fmt.Printf("init config failed, err: %s", err.Error()) + os.Exit(1) + } + + err = logger.Init(&config.Cfg().Logger) + if err != nil { + fmt.Printf("init logger failed, err: %s", err.Error()) + os.Exit(1) + } + + myglbs.Init() + + rpter := reporter.NewReporter(myglbs.AdvisorID, time.Second*time.Duration(config.Cfg().ReportIntervalSec)) + + taskMgr := task.NewManager(&rpter) + + mqSvr, err := execmq.NewServer(services.NewService(&taskMgr), &config.Cfg().RabbitMQ) + if err != nil { + logger.Fatalf("new advisor server failed, err: %s", err.Error()) + } + + mqSvr.OnError = func(err error) { + logger.Warnf("advisor server err: %s", err.Error()) + } + + // 启动服务 + go serveMQServer(mqSvr) + + go serveReporter(&rpter) + + forever := make(chan bool) + <-forever +} + +func serveMQServer(server *execmq.Server) { + logger.Info("start serving mq server") + + err := server.Serve() + if err != nil { + logger.Errorf("mq server stopped with error: %s", err.Error()) + } + + logger.Info("mq server stopped") +} + +func serveReporter(rpt *reporter.Reporter) { + logger.Info("start serving reporter") + + err := rpt.Serve() + if err != nil { + logger.Errorf("rpt stopped with error: %s", err.Error()) + } + + logger.Info("rpt stopped") +} diff --git a/common/pkgs/mq/advisor/apis.go b/common/pkgs/mq/advisor/apis.go new file mode 100644 index 0000000..d67646f --- /dev/null +++ b/common/pkgs/mq/advisor/apis.go @@ -0,0 +1,39 @@ +package advisor + +import ( + "gitlink.org.cn/cloudream/common/models" + "gitlink.org.cn/cloudream/common/pkgs/mq" +) + +// 获取调度方案 +var _ = Register(Service.StartGetScheduleScheme) + +type StartGetScheduleScheme struct { + // UserID int64 `json:"userID"` + // PackageID int64 `json:"packageID"` +} + +func NewStartGetScheduleScheme() StartGetScheduleScheme { + return StartGetScheduleScheme{ + // UserID: userID, + // PackageID: packageID, + } +} + +type StartGetScheduleSchemeResp struct { + TaskID string `json:"taskID"` +} + +func NewStartGetScheduleSchemeResp(taskID string) StartGetScheduleSchemeResp { + return StartGetScheduleSchemeResp{ + TaskID: taskID, + } +} + +func (c *Client) StartGetScheduleScheme(msg StartGetScheduleScheme, opts ...mq.RequestOption) (*StartGetScheduleSchemeResp, error) { + return mq.Request[StartGetScheduleSchemeResp](c.rabbitCli, msg, opts...) +} + +func init() { + mq.RegisterUnionType(models.ResourceDataTypeUnion) +} diff --git a/common/pkgs/mq/advisor/client.go b/common/pkgs/mq/advisor/client.go new file mode 100644 index 0000000..ea84058 --- /dev/null +++ b/common/pkgs/mq/advisor/client.go @@ -0,0 +1,59 @@ +package advisor + +import ( + "gitlink.org.cn/cloudream/common/pkgs/mq" + mymq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq" +) + +type Client struct { + rabbitCli *mq.RabbitMQClient +} + +func NewClient(cfg *mymq.Config) (*Client, error) { + rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), ServerQueueName, "") + if err != nil { + return nil, err + } + + return &Client{ + rabbitCli: rabbitCli, + }, nil +} + +func (c *Client) Close() { + c.rabbitCli.Close() +} + +type PoolClient struct { + *Client + owner *Pool +} + +func (c *PoolClient) Close() { + c.owner.Release(c) +} + +type Pool struct { + mqcfg *mymq.Config +} + +func NewPool(mqcfg *mymq.Config) *Pool { + return &Pool{ + mqcfg: mqcfg, + } +} +func (p *Pool) Acquire() (*PoolClient, error) { + cli, err := NewClient(p.mqcfg) + if err != nil { + return nil, err + } + + return &PoolClient{ + Client: cli, + owner: p, + }, nil +} + +func (p *Pool) Release(cli *PoolClient) { + cli.Client.Close() +} diff --git a/common/pkgs/mq/advisor/server.go b/common/pkgs/mq/advisor/server.go new file mode 100644 index 0000000..aa05d35 --- /dev/null +++ b/common/pkgs/mq/advisor/server.go @@ -0,0 +1,70 @@ +package advisor + +import ( + "gitlink.org.cn/cloudream/common/pkgs/mq" + mymq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq" +) + +const ( + ServerQueueName = "Scheduler-Collector" +) + +type Service interface { + StartGetScheduleScheme(msg *StartGetScheduleScheme) (*StartGetScheduleSchemeResp, *mq.CodeMessage) +} + +type Server struct { + service Service + rabbitSvr mq.RabbitMQServer + + OnError func(err error) +} + +func NewServer(svc Service, cfg *mymq.Config) (*Server, error) { + srv := &Server{ + service: svc, + } + + rabbitSvr, err := mq.NewRabbitMQServer( + cfg.MakeConnectingURL(), + ServerQueueName, + func(msg *mq.Message) (*mq.Message, error) { + return msgDispatcher.Handle(srv.service, msg) + }, + ) + if err != nil { + return nil, err + } + + srv.rabbitSvr = *rabbitSvr + + return srv, nil +} +func (s *Server) Stop() { + s.rabbitSvr.Close() +} + +func (s *Server) Serve() error { + return s.rabbitSvr.Serve() +} + +var msgDispatcher mq.MessageDispatcher = mq.NewMessageDispatcher() + +// Register 将Service中的一个接口函数作为指定类型消息的处理函数，同时会注册请求和响应的消息类型 +// TODO 需要约束：Service实现了TSvc接口 +func Register[TSvc any, TReq any, TResp any](svcFn func(svc TSvc, msg *TReq) (*TResp, *mq.CodeMessage)) any { + mq.AddServiceFn(&msgDispatcher, svcFn) + mq.RegisterMessage[TReq]() + mq.RegisterMessage[TResp]() + + return nil +} + +// RegisterNoReply 将Service中的一个*没有返回值的*接口函数作为指定类型消息的处理函数，同时会注册请求和响应的消息类型 +// TODO 需要约束：Service实现了TSvc接口 +func RegisterNoReply[TSvc any, TReq any](svcFn func(svc TSvc, msg *TReq)) any { + mq.AddNoRespServiceFn(&msgDispatcher, svcFn) + mq.RegisterMessage[TReq]() + + return nil +} diff --git a/common/pkgs/mq/advisor/task/task.go b/common/pkgs/mq/advisor/task/task.go new file mode 100644 index 0000000..27e34c1 --- /dev/null +++ b/common/pkgs/mq/advisor/task/task.go @@ -0,0 +1,73 @@ +package task + +type TaskStatus interface{} + +type TaskStatusConst interface { + TaskStatus | ScheduleTaskStatus | UploadImageTaskStatus +} + +type ScheduleTaskStatus struct { + Status string `json:"status"` + Error string `json:"error"` + PCMJobID int64 `json:"pcmJobID"` +} + +func NewScheduleTaskStatus(status string, err string, pcmJobID int64) ScheduleTaskStatus { + return ScheduleTaskStatus{ + Status: status, + Error: err, + PCMJobID: pcmJobID, + } +} + +type UploadImageTaskStatus struct { + Status string `json:"status"` + Error string `json:"error"` + ImageID int64 `json:"imageID"` +} + +func NewUploadImageTaskStatus(status string, err string, imageID int64) UploadImageTaskStatus { + return UploadImageTaskStatus{ + Status: status, + Error: err, + ImageID: imageID, + } +} + +type CacheMovePackageTaskStatus struct { + Status string `json:"status"` + Error string `json:"error"` +} + +func NewCacheMovePackageTaskStatus(status string, err string) CacheMovePackageTaskStatus { + return CacheMovePackageTaskStatus{ + Status: status, + Error: err, + } +} + +type CreatePackageTaskStatus struct { + Status string `json:"status"` + Error string `json:"error"` + PackageID int64 `json:"packageID"` +} + +func NewCreatePackageTaskStatus(status string, err string, packageID int64) CreatePackageTaskStatus { + return CreatePackageTaskStatus{ + Status: status, + Error: err, + PackageID: packageID, + } +} + +type LoadPackageTaskStatus struct { + Status string `json:"status"` + Error string `json:"error"` +} + +func NewLoadPackageTaskStatus(status string, err string) LoadPackageTaskStatus { + return LoadPackageTaskStatus{ + Status: status, + Error: err, + } +} diff --git a/common/pkgs/mq/manager/api.go b/common/pkgs/mq/manager/api.go index c7abecd..5055c27 100644 --- a/common/pkgs/mq/manager/api.go +++ b/common/pkgs/mq/manager/api.go @@ -2,32 +2,33 @@ package manager import ( "gitlink.org.cn/cloudream/common/pkgs/mq" + advtsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/advisor/task" exectsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor/task" ) // 接收executor上报的存活状态及任务执行情况 -var _ = Register(Service.ReportTaskStatus) +var _ = Register(Service.ReportExecutorTaskStatus) -type ReportTaskStatus struct { +type ReportExecutorTaskStatus struct { ExecutorID string `json:"executorID"` TaskStatus []ExecutorTaskStatus `json:"taskStatus"` } -type ReportTaskStatusResp struct { +type ReportExecutorTaskStatusResp struct { } type ExecutorTaskStatus struct { TaskID string Status exectsk.TaskStatus } -func NewReportTaskStatus(executorID string, taskStatus []ExecutorTaskStatus) ReportTaskStatus { - return ReportTaskStatus{ +func NewReportExecutorTaskStatus(executorID string, taskStatus []ExecutorTaskStatus) ReportExecutorTaskStatus { + return ReportExecutorTaskStatus{ ExecutorID: executorID, TaskStatus: taskStatus, } } -func NewReportTaskStatusResp() ReportTaskStatusResp { - return ReportTaskStatusResp{} +func NewReportExecutorTaskStatusResp() ReportExecutorTaskStatusResp { + return ReportExecutorTaskStatusResp{} } func NewExecutorTaskStatus[T exectsk.TaskStatusConst](taskID string, status T) ExecutorTaskStatus { return ExecutorTaskStatus{ @@ -35,6 +36,40 @@ func NewExecutorTaskStatus[T exectsk.TaskStatusConst](taskID string, status T) E Status: status, } } -func (c *Client) ReportTaskStatus(msg ReportTaskStatus, opts ...mq.RequestOption) (*ReportTaskStatusResp, error) { - return mq.Request[ReportTaskStatusResp](c.rabbitCli, msg, opts...) +func (c *Client) ReportExecutorTaskStatus(msg ReportExecutorTaskStatus, opts ...mq.RequestOption) (*ReportExecutorTaskStatusResp, error) { + return mq.Request[ReportExecutorTaskStatusResp](c.rabbitCli, msg, opts...) +} + +// 接收advisor上报的存活状态及任务执行情况 +var _ = Register(Service.ReportAdvisorTaskStatus) + +type ReportAdvisorTaskStatus struct { + AdvisorID string `json:"advisorID"` + TaskStatus []AdvisorTaskStatus `json:"taskStatus"` +} + +type ReportAdvisorTaskStatusResp struct { +} +type AdvisorTaskStatus struct { + TaskID string + Status advtsk.TaskStatus +} + +func NewReportAdvisorTaskStatus(advisorID string, taskStatus []AdvisorTaskStatus) ReportAdvisorTaskStatus { + return ReportAdvisorTaskStatus{ + AdvisorID: advisorID, + TaskStatus: taskStatus, + } +} +func NewReportAdvisorTaskStatusResp() ReportExecutorTaskStatusResp { + return ReportExecutorTaskStatusResp{} +} +func NewAdvisorTaskStatus[T exectsk.TaskStatusConst](taskID string, status T) AdvisorTaskStatus { + return AdvisorTaskStatus{ + TaskID: taskID, + Status: status, + } +} +func (c *Client) ReportAdvisorTaskStatus(msg ReportAdvisorTaskStatus, opts ...mq.RequestOption) (*ReportAdvisorTaskStatusResp, error) { + return mq.Request[ReportAdvisorTaskStatusResp](c.rabbitCli, msg, opts...) } diff --git a/common/pkgs/mq/manager/server.go b/common/pkgs/mq/manager/server.go index 36d3786..fb56991 100644 --- a/common/pkgs/mq/manager/server.go +++ b/common/pkgs/mq/manager/server.go @@ -10,7 +10,8 @@ const ( ) type Service interface { - ReportTaskStatus(msg *ReportTaskStatus) (*ReportTaskStatusResp, *mq.CodeMessage) + ReportExecutorTaskStatus(msg *ReportExecutorTaskStatus) (*ReportExecutorTaskStatusResp, *mq.CodeMessage) + ReportAdvisorTaskStatus(msg *ReportAdvisorTaskStatus) (*ReportAdvisorTaskStatusResp, *mq.CodeMessage) } type Server struct { diff --git a/executor/README.md b/executor/README.md index 3c6312d..3102f47 100644 --- a/executor/README.md +++ b/executor/README.md @@ -1,2 +1,2 @@ -# scheduler-dcontroller +# scheduler-executor diff --git a/executor/internal/reporter/reporter.go b/executor/internal/reporter/reporter.go index e7b619b..5658d26 100644 --- a/executor/internal/reporter/reporter.go +++ b/executor/internal/reporter/reporter.go @@ -70,7 +70,7 @@ func (r *Reporter) Serve() error { r.taskStatus = make(map[string]exectsk.TaskStatus) r.taskStatusLock.Unlock() - _, err := magCli.ReportTaskStatus(mgrmq.NewReportTaskStatus(r.executorID, taskStatus)) + _, err := magCli.ReportExecutorTaskStatus(mgrmq.NewReportExecutorTaskStatus(r.executorID, taskStatus)) if err != nil { logger.Warnf("reporting to manager: %s", err.Error()) diff --git a/executor/internal/task/pcm_scheduler_task.go b/executor/internal/task/pcm_schedule_task.go similarity index 100% rename from executor/internal/task/pcm_scheduler_task.go rename to executor/internal/task/pcm_schedule_task.go