From 03cc3fe9789f4de7faf335a4c05121cf65568bd1 Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:31:58 +0800 Subject: [PATCH 01/20] Update README.md Former-commit-id: 835b51a8d270bc7fa8fa8b7e4fba1194aec078e4 --- README.md | 46 +++++++++++++++++++++++----------------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 1903e134..3f96a77c 100644 --- a/README.md +++ b/README.md @@ -7,35 +7,35 @@ ## What is Jointcloud and PCM -  Jointcloud Cooperation Environment (JCCE) comprises Distributed Cloud Trading, Distributed Cloud Community, Distributed Cloud Supervision, and a Blockchain-based Distributed Accounting System. JCCE, with its supporting technologies related to information and value exchange, facilitates breaking the information asymmetry among participants in cloud collaboration. It creates a win-win opportunity for all involved parties and provides robust support for innovative business models in the cloud collaboration computing environment. +Jointcloud Cooperation Environment (JCCE) comprises Distributed Cloud Trading, Distributed Cloud Community, Distributed Cloud Supervision, and a Blockchain-based Distributed Accounting System. JCCE, with its supporting technologies related to information and value exchange, facilitates breaking the information asymmetry among participants in cloud collaboration. It creates a win-win opportunity for all involved parties and provides robust support for innovative business models in the cloud collaboration computing environment. -  The vision of the Peer Cooperation Mechanism (PCM) is to build a set of collaboration mechanisms in the cloud environment where all service providers and demand-side entities can autonomously participate, achieve peer-to-peer transactions, and establish efficient connections and invocations in a non-intrusive manner. PCM focuses more on collaboration among clouds rather than individual clouds themselves. +The vision of the Peer Cooperation Mechanism (PCM) is to build a set of collaboration mechanisms in the cloud environment where all service providers and demand-side entities can autonomously participate, achieve peer-to-peer transactions, and establish efficient connections and invocations in a non-intrusive manner. PCM focuses more on collaboration among clouds rather than individual clouds themselves. -  PCM is built upon a standardized software-defined framework. Cloud service providers with physical cloud resources can autonomously define PCM rule frameworks. Other entities adhering to these rule frameworks can then participate in collaborative transactions. Subsequently, tailored virtual private clouds can be created for domain-specific applications, meeting the diverse requirements of various cloud participants, such as central, edge, and terminal infrastructure, enabling them to directly participate in collaboration and transactions. +PCM is built upon a standardized software-defined framework. Cloud service providers with physical cloud resources can autonomously define PCM rule frameworks. Other entities adhering to these rule frameworks can then participate in collaborative transactions. Subsequently, tailored virtual private clouds can be created for domain-specific applications, meeting the diverse requirements of various cloud participants, such as central, edge, and terminal infrastructure, enabling them to directly participate in collaboration and transactions. ## Real-world Issues Addressed by PCM: -- **Performance Improvement**: +- **Performance Improvement** Developers in the cloud collaboration environment experience performance enhancement upon joining the PCM framework. They can generate code for the internal functionality structure and most specification definitions using descriptive language, allowing them to focus on business logic development without concerning themselves with underlying management functions. The framework offers features such as microservices management, multi-language code generation, and model bridging, reducing coding workload and entry barriers while improving efficiency. -- **Platform Lock-in Resolution**: +- **Platform Lock-in Resolution** The PCM effectively addresses platform lock-in issues through multi-cloud adaptation, standardized interfaces, and abstraction layers. This enables cross-platform operations, deployment, and interaction. Standardized interfaces simplify the work for developers, lowering the risk of platform lock-in. -- **Reduced Code Development Complexity**: +- **Reduced Code Development Complexity** The PCM development framework lowers the complexity of development by providing structured development patterns, ready-made components, and documentation support. Developers of different skill levels can collaborate more fairly. The framework can save approximately 50% of development time, with junior programmers completing tasks that originally required one person-month in about 15 person-days. Features such as automatic generation tools, code templates, and component reuse allow developers to focus more on business logic implementation. There is no need for additional training of advanced developers, saving time, and reducing labor costs while improving the return on investment. - **Reduced Code Development Workload**: -　　 The PCM framework offers automation features and code generation tools, reducing manual code writing and improving development speed and code quality. With the framework's auto-generation tools, over half of the code is generated, achieving a low code rate of approximately 63%. Developers primarily work on writing descriptive language files to generate the basic structure and then adjust and write logic functions. This enables teams to deliver products more quickly, iterate based on business optimization and user feedback, and be more agile. +The PCM framework offers automation features and code generation tools, reducing manual code writing and improving development speed and code quality. With the framework's auto-generation tools, over half of the code is generated, achieving a low code rate of approximately 63%. Developers primarily work on writing descriptive language files to generate the basic structure and then adjust and write logic functions. This enables teams to deliver products more quickly, iterate based on business optimization and user feedback, and be more agile. ## Architecture -  The Coordinator is the core component of the framework, providing heterogeneous abstraction for different Participant-side technology stacks. The framework, oriented towards the user side, primarily involves two major functionalities: resource changes (task submission) and centralized display. After task submission, PCM achieves dynamic task flow through a scheduling center deployed on cloud computing, intelligent computing, and supercomputing clouds. In the centralized display section, PCM mainly collects and aggregates information about resources and tasks from multiple Participant services through Tracker and actively reported by the Participant side. It provides users with a standardized unified interface. Users can view the information provided by PCM on the frontend page or directly access data through the interface. +The Coordinator is the core component of the framework, providing heterogeneous abstraction for different Participant-side technology stacks. The framework, oriented towards the user side, primarily involves two major functionalities: resource changes (task submission) and centralized display. After task submission, PCM achieves dynamic task flow through a scheduling center deployed on cloud computing, intelligent computing, and supercomputing clouds. In the centralized display section, PCM mainly collects and aggregates information about resources and tasks from multiple Participant services through Tracker and actively reported by the Participant side. It provides users with a standardized unified interface. Users can view the information provided by PCM on the frontend page or directly access data through the interface. ![PCM架构](docs/images/arch-eng.png) ## PCM deploy @@ -43,22 +43,22 @@ The development environment for PCM requires the installation of Go version 1.18 or above. Please refer to the following instructions to locally start the Kubernetes-related services. The command to fetch the project is as follows: -``` +```bash git clone https://gitlink.org.cn/JointCloud/pcm-coordinator.git ``` -  After executing the following command, the Kubernetes RPC service will be launched locally. For its specific functionalities, please refer to the description in the architecture design mentioned earlier. -``` +After executing the following command, the Kubernetes RPC service will be launched locally. For its specific functionalities, please refer to the description in the architecture design mentioned earlier. +``` bash # get required packages go mod tidy # run rpc service go run pcm-coordinator/rpc/pcmcore.go -``` +``` #### coordinator-api -  The template for the configuration content of the API service is as follows, where the config information for each Participant (P-side) can be configured as needed based on the actual situation. -``` +The template for the configuration content of the API service is as follows, where the config information for each Participant (P-side) can be configured as needed based on the actual situation. +``` bash Name: pcm.core.api Host: 0.0.0.0 Port: 8999 @@ -77,22 +77,22 @@ THRpcConf: #modelarts rpc ModelArtsRpcConf: Endpoints: - 127.0.0.1:2002NonBlock: true -``` -  After executing the following command, the Kubernetes API service will be launched locally. Once the service is started, users can make HTTP requests to its interfaces for various functional calls. -``` +``` +After executing the following command, the Kubernetes API service will be launched locally. Once the service is started, users can make HTTP requests to its interfaces for various functional calls. +``` bash # get required packages go mod tidy # run rpc service go run pcm-coordinator/api/pcm.go -``` +``` ## Upcoming Plans -- Pluginization of Scheduling Algorithms and Definition of Basic Resource Templates -- Fundamental Definition of Resource Operation Types -- Rapid Development Mode -- Implementation of First-level Scheduling +- Pluginization of Scheduling Algorithms and Definition of Basic Resource Templates. +- Fundamental Definition of Resource Operation Types. +- Rapid Development Mode. +- Implementation of First-level Scheduling. ## Join＆Contribute -  We look forward to your opinions and contributions. Welcome all friends to provide corrections and improvements to the project, collectively building an efficient and stable cloud collaboration mechanism. \ No newline at end of file +We look forward to your opinions and contributions. Welcome all friends to provide corrections and improvements to the project, collectively building an efficient and stable cloud collaboration mechanism. \ No newline at end of file From c15cc7254aab1ed67c90beb564691b791d466979 Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:33:42 +0800 Subject: [PATCH 02/20] Update README.md Former-commit-id: 55b9855d043a04b6c76e1e7a973ab0d80903480c --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 3f96a77c..320bfeec 100644 --- a/README.md +++ b/README.md @@ -27,7 +27,7 @@ PCM is built upon a standardized software-defined framework. Cloud service provi - **Reduced Code Development Complexity** - The PCM development framework lowers the complexity of development by providing structured development patterns, ready-made components, and documentation support. Developers of different skill levels can collaborate more fairly. The framework can save approximately 50% of development time, with junior programmers completing tasks that originally required one person-month in about 15 person-days. Features such as automatic generation tools, code templates, and component reuse allow developers to focus more on business logic implementation. There is no need for additional training of advanced developers, saving time, and reducing labor costs while improving the return on investment. +The PCM development framework lowers the complexity of development by providing structured development patterns, ready-made components, and documentation support. Developers of different skill levels can collaborate more fairly. The framework can save approximately 50% of development time, with junior programmers completing tasks that originally required one person-month in about 15 person-days. Features such as automatic generation tools, code templates, and component reuse allow developers to focus more on business logic implementation. There is no need for additional training of advanced developers, saving time, and reducing labor costs while improving the return on investment. - **Reduced Code Development Workload**: From 4838de910c616bc005c5060a872eb0c6c8c0e09d Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:34:04 +0800 Subject: [PATCH 03/20] Update README.md Former-commit-id: c595a58fd98428f428139648d52bb30793e40754 --- README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/README.md b/README.md index 320bfeec..f684f986 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,6 @@ PCM is built upon a standardized software-defined framework. Cloud service provi - **Reduced Code Development Complexity** - The PCM development framework lowers the complexity of development by providing structured development patterns, ready-made components, and documentation support. Developers of different skill levels can collaborate more fairly. The framework can save approximately 50% of development time, with junior programmers completing tasks that originally required one person-month in about 15 person-days. Features such as automatic generation tools, code templates, and component reuse allow developers to focus more on business logic implementation. There is no need for additional training of advanced developers, saving time, and reducing labor costs while improving the return on investment. From eecb88541386a1dc895b1c2a782f2e01ad7b95c4 Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:34:49 +0800 Subject: [PATCH 04/20] Update README.md Former-commit-id: 71476e27b72c3e4a868d4245dc5000feeaeb3417 --- README.md | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index f684f986..fe692cab 100644 --- a/README.md +++ b/README.md @@ -16,13 +16,11 @@ PCM is built upon a standardized software-defined framework. Cloud service provi ## Real-world Issues Addressed by PCM: - **Performance Improvement** - - Developers in the cloud collaboration environment experience performance enhancement upon joining the PCM framework. They can generate code for the internal functionality structure and most specification definitions using descriptive language, allowing them to focus on business logic development without concerning themselves with underlying management functions. The framework offers features such as microservices management, multi-language code generation, and model bridging, reducing coding workload and entry barriers while improving efficiency. +Developers in the cloud collaboration environment experience performance enhancement upon joining the PCM framework. They can generate code for the internal functionality structure and most specification definitions using descriptive language, allowing them to focus on business logic development without concerning themselves with underlying management functions. The framework offers features such as microservices management, multi-language code generation, and model bridging, reducing coding workload and entry barriers while improving efficiency. - **Platform Lock-in Resolution** - - The PCM effectively addresses platform lock-in issues through multi-cloud adaptation, standardized interfaces, and abstraction layers. This enables cross-platform operations, deployment, and interaction. Standardized interfaces simplify the work for developers, lowering the risk of platform lock-in. +The PCM effectively addresses platform lock-in issues through multi-cloud adaptation, standardized interfaces, and abstraction layers. This enables cross-platform operations, deployment, and interaction. Standardized interfaces simplify the work for developers, lowering the risk of platform lock-in. - **Reduced Code Development Complexity** From 4fa42ed43337c2ce5f67c763c8a9becfbdcc1319 Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:35:54 +0800 Subject: [PATCH 05/20] Update README.md Former-commit-id: 406286681bcb8a6715e82da292e354df7b82944f --- README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index fe692cab..01192752 100644 --- a/README.md +++ b/README.md @@ -15,19 +15,19 @@ PCM is built upon a standardized software-defined framework. Cloud service provi ## Real-world Issues Addressed by PCM: -- **Performance Improvement** +- **Performance Improvement** Developers in the cloud collaboration environment experience performance enhancement upon joining the PCM framework. They can generate code for the internal functionality structure and most specification definitions using descriptive language, allowing them to focus on business logic development without concerning themselves with underlying management functions. The framework offers features such as microservices management, multi-language code generation, and model bridging, reducing coding workload and entry barriers while improving efficiency. -- **Platform Lock-in Resolution** +- **Platform Lock-in Resolution** The PCM effectively addresses platform lock-in issues through multi-cloud adaptation, standardized interfaces, and abstraction layers. This enables cross-platform operations, deployment, and interaction. Standardized interfaces simplify the work for developers, lowering the risk of platform lock-in. -- **Reduced Code Development Complexity** +- **Reduced Code Development Complexity** The PCM development framework lowers the complexity of development by providing structured development patterns, ready-made components, and documentation support. Developers of different skill levels can collaborate more fairly. The framework can save approximately 50% of development time, with junior programmers completing tasks that originally required one person-month in about 15 person-days. Features such as automatic generation tools, code templates, and component reuse allow developers to focus more on business logic implementation. There is no need for additional training of advanced developers, saving time, and reducing labor costs while improving the return on investment. -- **Reduced Code Development Workload**: +- **Reduced Code Development Workload** The PCM framework offers automation features and code generation tools, reducing manual code writing and improving development speed and code quality. With the framework's auto-generation tools, over half of the code is generated, achieving a low code rate of approximately 63%. Developers primarily work on writing descriptive language files to generate the basic structure and then adjust and write logic functions. This enables teams to deliver products more quickly, iterate based on business optimization and user feedback, and be more agile. From 92d2b31f76a94b59b7172d355748819d76862e05 Mon Sep 17 00:00:00 2001 From: Tuberrr Date: Mon, 29 Apr 2024 15:39:41 +0800 Subject: [PATCH 06/20] Update README.md Former-commit-id: daf75cec37db83641da53e436d1d9afe51ccb7f1 --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 01192752..2b05037d 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ -![PCM](https://www.gitlink.org.cn/images/avatars/Organization/123822?t=1689062058) +PCM +

Docs | 简体中文 | From 0e155cb6573d1569a025c7bdcc247211beac7f42 Mon Sep 17 00:00:00 2001 From: tzwang Date: Thu, 9 May 2024 16:00:40 +0800 Subject: [PATCH 07/20] added algorithmscode api Former-commit-id: 60b3ed6bb4c24d801afeec3b4cee108788cd2bea --- api/desc/pcm.api | 6 ++++ api/desc/schedule/pcm-schedule.api | 29 +++++++++++++++++++ .../logic/ai/getcenteroverviewlogic.go | 10 +++++-- .../logic/ai/getcentertasklistlogic.go | 9 ++++-- .../scheduler/service/collector/collector.go | 4 +-- api/internal/storeLink/modelarts.go | 4 +-- api/internal/storeLink/octopus.go | 4 +-- api/internal/storeLink/shuguangai.go | 27 +++++++++++++++-- 8 files changed, 80 insertions(+), 13 deletions(-) diff --git a/api/desc/pcm.api b/api/desc/pcm.api index a124833c..2515c1e1 100644 --- a/api/desc/pcm.api +++ b/api/desc/pcm.api @@ -954,6 +954,12 @@ service pcm { @handler ScheduleGetOverviewHandler post /schedule/getOverview returns (ScheduleOverviewResp) + + @handler DownloadAlgothmCodeHandler + get /schedule/getDownloadAlgothmCode (DownloadAlgorithmCodeReq) returns (DownloadAlgorithmCodeResp) + + @handler UploadAlgothmCodeHandler + post /schedule/getDownloadAlgothmCode (UploadAlgorithmCodeReq) returns (UploadAlgorithmCodeResp) } @server( diff --git a/api/desc/schedule/pcm-schedule.api b/api/desc/schedule/pcm-schedule.api index e1892770..0611dc4d 100644 --- a/api/desc/schedule/pcm-schedule.api +++ b/api/desc/schedule/pcm-schedule.api @@ -100,4 +100,33 @@ type ( StartTime string `json:"startTime,omitempty" db:"start_time"` EndTime string `json:"endTime,omitempty" db:"end_time"` } + + DownloadAlgorithmCodeReq { + AdapterId string `form:"adapterId"` + ClusterId string `form:"clusterId"` + ResourceType string `form:"resourceType"` + Card string `form:"card"` + TaskType string `form:"taskType"` + Dataset string `form:"dataset"` + Algorithm string `form:"algorithm"` + Code string `form:"code"` + } + + DownloadAlgorithmCodeResp { + Code string `json:"algorithms"` + } + + UploadAlgorithmCodeReq { + AdapterId string `json:"adapterId"` + ClusterId string `json:"clusterId"` + ResourceType string `json:"resourceType"` + Card string `json:"card"` + TaskType string `json:"taskType"` + Dataset string `json:"dataset"` + Algorithm string `json:"algorithm"` + Code string `json:"code"` + } + + UploadAlgorithmCodeResp { + } ) \ No newline at end of file diff --git a/api/internal/logic/ai/getcenteroverviewlogic.go b/api/internal/logic/ai/getcenteroverviewlogic.go index 94d44cea..d2669709 100644 --- a/api/internal/logic/ai/getcenteroverviewlogic.go +++ b/api/internal/logic/ai/getcenteroverviewlogic.go @@ -8,6 +8,7 @@ import ( "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models" "strconv" "sync" + "time" ) type GetCenterOverviewLogic struct { @@ -71,9 +72,14 @@ func (l *GetCenterOverviewLogic) GetCenterOverview() (resp *types.CenterOverview } resp.CardNum = cardNum resp.PowerInTops = totalTops - <-ch - return resp, nil + select { + case _ = <-ch: + return resp, nil + case <-time.After(2 * time.Second): + return resp, nil + } + } func (l *GetCenterOverviewLogic) updateClusterResource(mu *sync.RWMutex, ch chan<- struct{}, list []*types.AdapterInfo) { diff --git a/api/internal/logic/ai/getcentertasklistlogic.go b/api/internal/logic/ai/getcentertasklistlogic.go index ebca4dc4..ff1d2883 100644 --- a/api/internal/logic/ai/getcentertasklistlogic.go +++ b/api/internal/logic/ai/getcentertasklistlogic.go @@ -73,9 +73,14 @@ func (l *GetCenterTaskListLogic) GetCenterTaskList() (resp *types.CenterTaskList resp.List = append(resp.List, t) } } - <-ch - return resp, nil + select { + case _ = <-ch: + return resp, nil + case <-time.After(1 * time.Second): + return resp, nil + } + } func (l *GetCenterTaskListLogic) updateAiTaskStatus(mu *sync.RWMutex, ch chan<- struct{}, list []*types.AdapterInfo) { diff --git a/api/internal/scheduler/service/collector/collector.go b/api/internal/scheduler/service/collector/collector.go index 96ddc815..453d710c 100644 --- a/api/internal/scheduler/service/collector/collector.go +++ b/api/internal/scheduler/service/collector/collector.go @@ -8,8 +8,8 @@ type AiCollector interface { GetAlgorithms(ctx context.Context) ([]*Algorithm, error) GetTrainingTaskLog(ctx context.Context, taskId string, instanceNum string) (string, error) GetTrainingTask(ctx context.Context, taskId string) (*Task, error) - DownloadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string) (string, error) - UploadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string, code string) error + DownloadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string) (string, error) + UploadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string, code string) error } type ResourceStats struct { diff --git a/api/internal/storeLink/modelarts.go b/api/internal/storeLink/modelarts.go index 1ae255f2..7bb6db2d 100644 --- a/api/internal/storeLink/modelarts.go +++ b/api/internal/storeLink/modelarts.go @@ -162,11 +162,11 @@ func (m *ModelArtsLink) GetAlgorithms(ctx context.Context) ([]*collector.Algorit return nil, nil } -func (m *ModelArtsLink) DownloadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string) (string, error) { +func (m *ModelArtsLink) DownloadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string) (string, error) { return "", nil } -func (m *ModelArtsLink) UploadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string, code string) error { +func (m *ModelArtsLink) UploadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string, code string) error { return nil } diff --git a/api/internal/storeLink/octopus.go b/api/internal/storeLink/octopus.go index f72f63ab..8e1f7af0 100644 --- a/api/internal/storeLink/octopus.go +++ b/api/internal/storeLink/octopus.go @@ -339,11 +339,11 @@ func (o *OctopusLink) GetAlgorithms(ctx context.Context) ([]*collector.Algorithm return algorithms, nil } -func (o *OctopusLink) DownloadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string) (string, error) { +func (o *OctopusLink) DownloadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string) (string, error) { return "", nil } -func (o *OctopusLink) UploadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string, code string) error { +func (o *OctopusLink) UploadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string, code string) error { return nil } diff --git a/api/internal/storeLink/shuguangai.go b/api/internal/storeLink/shuguangai.go index 4dbd8fcc..4f783357 100644 --- a/api/internal/storeLink/shuguangai.go +++ b/api/internal/storeLink/shuguangai.go @@ -447,11 +447,32 @@ func (s *ShuguangAi) GetAlgorithms(ctx context.Context) ([]*collector.Algorithm, return algorithms, nil } -func (s *ShuguangAi) DownloadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string) (string, error) { - return "", nil +func (s *ShuguangAi) DownloadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string) (string, error) { + algoName := dataset + DASH + algorithm + req := &hpcAC.GetFileReq{ + Path: ALGORITHM_DIR + FORWARD_SLASH + taskType + FORWARD_SLASH + algoName + FORWARD_SLASH + TRAIN_FILE, + } + resp, err := s.aCRpc.GetFile(ctx, req) + if err != nil { + return "", err + } + + return resp.Content, nil } -func (s *ShuguangAi) UploadAlgorithmCode(ctx context.Context, resourceType string, taskType string, dataset string, algorithm string, code string) error { +func (s *ShuguangAi) UploadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string, code string) error { + algoName := dataset + DASH + algorithm + req := &hpcAC.UploadFileReq{ + Path: ALGORITHM_DIR + FORWARD_SLASH + taskType + FORWARD_SLASH + algoName + FORWARD_SLASH, + Cover: "cover", + File: code, + } + + _, err := s.aCRpc.UploadFile(ctx, req) + if err != nil { + return err + } + return nil } From 9ec3a048b0584b87528d906cf61b7b161de110ec Mon Sep 17 00:00:00 2001 From: tzwang Date: Thu, 9 May 2024 16:11:29 +0800 Subject: [PATCH 08/20] gen algorithmcode proto Former-commit-id: 5efb7dd3acce89cd06ae014f4e84fc02cbf5a5a6 --- api/internal/handler/routes.go | 17 ++++++++++++++++- api/internal/types/types.go | 29 +++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/api/internal/handler/routes.go b/api/internal/handler/routes.go index bedd9c57..c558141a 100644 --- a/api/internal/handler/routes.go +++ b/api/internal/handler/routes.go @@ -1190,6 +1190,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/schedule/ai/getAlgorithms/:adapterId/:resourceType/:taskType/:dataset", Handler: schedule.ScheduleGetAlgorithmsHandler(serverCtx), }, + { + Method: http.MethodGet, + Path: "/schedule/ai/getJobLog/:adapterId/:clusterId/:taskId/:instanceNum", + Handler: schedule.ScheduleGetAiJobLogLogHandler(serverCtx), + }, { Method: http.MethodPost, Path: "/schedule/submit", @@ -1200,6 +1205,16 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/schedule/getOverview", Handler: schedule.ScheduleGetOverviewHandler(serverCtx), }, + { + Method: http.MethodGet, + Path: "/schedule/getDownloadAlgothmCode", + Handler: schedule.DownloadAlgothmCodeHandler(serverCtx), + }, + { + Method: http.MethodPost, + Path: "/schedule/getDownloadAlgothmCode", + Handler: schedule.UploadAlgothmCodeHandler(serverCtx), + }, }, rest.WithPrefix("/pcm/v1"), ) @@ -1294,7 +1309,7 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { }, { Method: http.MethodPost, - Path: "/core/syncClusterAlert", + Path: "/monitoring/syncClusterAlert", Handler: monitoring.SyncClusterAlertHandler(serverCtx), }, { diff --git a/api/internal/types/types.go b/api/internal/types/types.go index 0a905601..887ef866 100644 --- a/api/internal/types/types.go +++ b/api/internal/types/types.go @@ -5676,6 +5676,35 @@ type AiTaskDb struct { EndTime string `json:"endTime,omitempty" db:"end_time"` } +type DownloadAlgorithmCodeReq struct { + AdapterId string `form:"adapterId"` + ClusterId string `form:"clusterId"` + ResourceType string `form:"resourceType"` + Card string `form:"card"` + TaskType string `form:"taskType"` + Dataset string `form:"dataset"` + Algorithm string `form:"algorithm"` + Code string `form:"code"` +} + +type DownloadAlgorithmCodeResp struct { + Code string `json:"algorithms"` +} + +type UploadAlgorithmCodeReq struct { + AdapterId string `json:"adapterId"` + ClusterId string `json:"clusterId"` + ResourceType string `json:"resourceType"` + Card string `json:"card"` + TaskType string `json:"taskType"` + Dataset string `json:"dataset"` + Algorithm string `json:"algorithm"` + Code string `json:"code"` +} + +type UploadAlgorithmCodeResp struct { +} + type CreateAlertRuleReq struct { CLusterId string `json:"clusterId"` ClusterName string `json:"clusterName"` From 4849b9385325817258d6562a472ebd7d2b8680b2 Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Thu, 9 May 2024 18:37:24 +0800 Subject: [PATCH 09/20] table name changed Former-commit-id: 5776bc02745e66782bf7f536b8a239a52095d78f --- api/internal/logic/core/pushtaskinfologic.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/api/internal/logic/core/pushtaskinfologic.go b/api/internal/logic/core/pushtaskinfologic.go index 5d86bca2..664a8df9 100644 --- a/api/internal/logic/core/pushtaskinfologic.go +++ b/api/internal/logic/core/pushtaskinfologic.go @@ -51,7 +51,7 @@ func (l *PushTaskInfoLogic) PushTaskInfo(req *clientCore.PushTaskInfoReq) (*clie } case 1: for _, aiInfo := range req.AiInfoList { - l.svcCtx.DbEngin.Exec("update ai set status = ?,start_time = ?,project_id = ?,job_id = ? where participant_id = ? and task_id = ? and name = ?", + l.svcCtx.DbEngin.Exec("update task_ai set status = ?,start_time = ?,project_id = ?,job_id = ? where participant_id = ? and task_id = ? and name = ?", aiInfo.Status, aiInfo.StartTime, aiInfo.ProjectId, aiInfo.JobId, req.AdapterId, aiInfo.TaskId, aiInfo.Name) syncTask(l.svcCtx.DbEngin, aiInfo.TaskId) } @@ -69,7 +69,7 @@ func (l *PushTaskInfoLogic) PushTaskInfo(req *clientCore.PushTaskInfoReq) (*clie func syncTask(gorm *gorm.DB, taskId int64) { var allStatus string - tx := gorm.Raw("SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.status) ,GROUP_CONCAT(DISTINCT a.status) ,GROUP_CONCAT(DISTINCT c.status))as status from task t left join hpc h on t.id = h.task_id left join task_cloud c on t.id = c.task_id left join ai a on t.id = a.task_id where t.id = ?", taskId).Scan(&allStatus) + tx := gorm.Raw("SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.status) ,GROUP_CONCAT(DISTINCT a.status) ,GROUP_CONCAT(DISTINCT c.status))as status from task t left join task_hpc h on t.id = h.task_id left join task_cloud c on t.id = c.task_id left join task_ai a on t.id = a.task_id where t.id = ?", taskId).Scan(&allStatus) if tx.Error != nil { logx.Error(tx.Error) } From fb957526e518cde762412ee83ef10538a1ebb322 Mon Sep 17 00:00:00 2001 From: tzwang Date: Thu, 9 May 2024 18:54:25 +0800 Subject: [PATCH 10/20] added getComputeCards api Former-commit-id: 78b5fa73e9106a813df8d05e57c1e5dfd0a219a4 --- api/desc/pcm.api | 3 +++ api/desc/schedule/pcm-schedule.api | 9 +++++++++ api/internal/logic/ai/getcentertasklistlogic.go | 2 +- 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/api/desc/pcm.api b/api/desc/pcm.api index 8a3ac032..73c6577d 100644 --- a/api/desc/pcm.api +++ b/api/desc/pcm.api @@ -972,6 +972,9 @@ service pcm { @handler UploadAlgothmCodeHandler post /schedule/getDownloadAlgothmCode (UploadAlgorithmCodeReq) returns (UploadAlgorithmCodeResp) + + @handler GetComputeCardsByClusterHandler + get /schedule/getComputeCardsByCluster/:adapterId/:clusterId (GetComputeCardsByClusterReq) returns (GetComputeCardsByClusterResp) } @server( diff --git a/api/desc/schedule/pcm-schedule.api b/api/desc/schedule/pcm-schedule.api index 0611dc4d..776851db 100644 --- a/api/desc/schedule/pcm-schedule.api +++ b/api/desc/schedule/pcm-schedule.api @@ -129,4 +129,13 @@ type ( UploadAlgorithmCodeResp { } + + GetComputeCardsByClusterReq { + AdapterId string `path:"adapterId"` + ClusterId string `path:"clusterId"` + } + + GetComputeCardsByClusterResp { + Cards []string `json:"cards"` + } ) \ No newline at end of file diff --git a/api/internal/logic/ai/getcentertasklistlogic.go b/api/internal/logic/ai/getcentertasklistlogic.go index ff1d2883..edf3d1b4 100644 --- a/api/internal/logic/ai/getcentertasklistlogic.go +++ b/api/internal/logic/ai/getcentertasklistlogic.go @@ -77,7 +77,7 @@ func (l *GetCenterTaskListLogic) GetCenterTaskList() (resp *types.CenterTaskList select { case _ = <-ch: return resp, nil - case <-time.After(1 * time.Second): + case <-time.After(2 * time.Second): return resp, nil } From 8d8d6a9822ab12fc0cf9ff9e84faa8b2a18dd7e4 Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Thu, 9 May 2024 19:05:16 +0800 Subject: [PATCH 11/20] push notice when new hpc job submitted Former-commit-id: 459845bff6477ce1a0b43cba3eac68aa2c7d1223 --- api/internal/logic/hpc/commithpctasklogic.go | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/api/internal/logic/hpc/commithpctasklogic.go b/api/internal/logic/hpc/commithpctasklogic.go index 636400df..b763a068 100644 --- a/api/internal/logic/hpc/commithpctasklogic.go +++ b/api/internal/logic/hpc/commithpctasklogic.go @@ -2,10 +2,12 @@ package hpc import ( "context" + clientCore "gitlink.org.cn/JointCloud/pcm-coordinator/api/client" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models" "k8s.io/apimachinery/pkg/util/json" "math/rand" + "strconv" "time" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" @@ -88,6 +90,21 @@ func (l *CommitHpcTaskLogic) CommitHpcTask(req *types.CommitHpcTaskReq) (resp *t if tx.Error != nil { return nil, tx.Error } + adapterId, _ := strconv.ParseUint(req.AdapterId, 10, 64) + adapterName := "" + tx.Table("t_adapter").Select("name").Where("id=?", adapterId).Find(&adapterName) + noticeInfo := clientCore.NoticeInfo{ + AdapterId: int64(adapterId), + AdapterName: adapterName, + NoticeType: "create", + TaskName: req.Name, + Incident: "任务创建中", + CreatedTime: time.Now(), + } + result := l.svcCtx.DbEngin.Table("t_notice").Create(¬iceInfo) + if result.Error != nil { + logx.Errorf("Task creation failure, err: %v", result.Error) + } // todo mq task manage //reqMessage, err := json.Marshal(mqInfo) //if err != nil { From 6bc14cfabd0e0e547b299a5f77a34c328e038892 Mon Sep 17 00:00:00 2001 From: tzwang Date: Thu, 9 May 2024 19:11:44 +0800 Subject: [PATCH 12/20] updated protobuf Former-commit-id: 6e3382beb8290c1c9602ca197c3781ce8a80a296 --- api/internal/handler/routes.go | 5 ++++ .../schedule/downloadalgothmcodehandler.go | 28 +++++++++++++++++++ .../getcomputecardsbyclusterhandler.go | 28 +++++++++++++++++++ .../schedule/uploadalgothmcodehandler.go | 28 +++++++++++++++++++ api/internal/types/types.go | 9 ++++++ 5 files changed, 98 insertions(+) create mode 100644 api/internal/handler/schedule/downloadalgothmcodehandler.go create mode 100644 api/internal/handler/schedule/getcomputecardsbyclusterhandler.go create mode 100644 api/internal/handler/schedule/uploadalgothmcodehandler.go diff --git a/api/internal/handler/routes.go b/api/internal/handler/routes.go index c558141a..0a2e86a0 100644 --- a/api/internal/handler/routes.go +++ b/api/internal/handler/routes.go @@ -1215,6 +1215,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/schedule/getDownloadAlgothmCode", Handler: schedule.UploadAlgothmCodeHandler(serverCtx), }, + { + Method: http.MethodGet, + Path: "/schedule/getComputeCardsByCluster/:adapterId/:clusterId", + Handler: schedule.GetComputeCardsByClusterHandler(serverCtx), + }, }, rest.WithPrefix("/pcm/v1"), ) diff --git a/api/internal/handler/schedule/downloadalgothmcodehandler.go b/api/internal/handler/schedule/downloadalgothmcodehandler.go new file mode 100644 index 00000000..14207bba --- /dev/null +++ b/api/internal/handler/schedule/downloadalgothmcodehandler.go @@ -0,0 +1,28 @@ +package schedule + +import ( + "net/http" + + "github.com/zeromicro/go-zero/rest/httpx" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/schedule" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" +) + +func DownloadAlgothmCodeHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req types.DownloadAlgorithmCodeReq + if err := httpx.Parse(r, &req); err != nil { + httpx.ErrorCtx(r.Context(), w, err) + return + } + + l := schedule.NewDownloadAlgothmCodeLogic(r.Context(), svcCtx) + resp, err := l.DownloadAlgothmCode(&req) + if err != nil { + httpx.ErrorCtx(r.Context(), w, err) + } else { + httpx.OkJsonCtx(r.Context(), w, resp) + } + } +} diff --git a/api/internal/handler/schedule/getcomputecardsbyclusterhandler.go b/api/internal/handler/schedule/getcomputecardsbyclusterhandler.go new file mode 100644 index 00000000..2c4393e1 --- /dev/null +++ b/api/internal/handler/schedule/getcomputecardsbyclusterhandler.go @@ -0,0 +1,28 @@ +package schedule + +import ( + "net/http" + + "github.com/zeromicro/go-zero/rest/httpx" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/schedule" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" +) + +func GetComputeCardsByClusterHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req types.GetComputeCardsByClusterReq + if err := httpx.Parse(r, &req); err != nil { + httpx.ErrorCtx(r.Context(), w, err) + return + } + + l := schedule.NewGetComputeCardsByClusterLogic(r.Context(), svcCtx) + resp, err := l.GetComputeCardsByCluster(&req) + if err != nil { + httpx.ErrorCtx(r.Context(), w, err) + } else { + httpx.OkJsonCtx(r.Context(), w, resp) + } + } +} diff --git a/api/internal/handler/schedule/uploadalgothmcodehandler.go b/api/internal/handler/schedule/uploadalgothmcodehandler.go new file mode 100644 index 00000000..bfbd05d1 --- /dev/null +++ b/api/internal/handler/schedule/uploadalgothmcodehandler.go @@ -0,0 +1,28 @@ +package schedule + +import ( + "net/http" + + "github.com/zeromicro/go-zero/rest/httpx" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/schedule" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" +) + +func UploadAlgothmCodeHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req types.UploadAlgorithmCodeReq + if err := httpx.Parse(r, &req); err != nil { + httpx.ErrorCtx(r.Context(), w, err) + return + } + + l := schedule.NewUploadAlgothmCodeLogic(r.Context(), svcCtx) + resp, err := l.UploadAlgothmCode(&req) + if err != nil { + httpx.ErrorCtx(r.Context(), w, err) + } else { + httpx.OkJsonCtx(r.Context(), w, resp) + } + } +} diff --git a/api/internal/types/types.go b/api/internal/types/types.go index 887ef866..4520f873 100644 --- a/api/internal/types/types.go +++ b/api/internal/types/types.go @@ -5705,6 +5705,15 @@ type UploadAlgorithmCodeReq struct { type UploadAlgorithmCodeResp struct { } +type GetComputeCardsByClusterReq struct { + AdapterId string `path:"adapterId"` + ClusterId string `path:"clusterId"` +} + +type GetComputeCardsByClusterResp struct { + Cards []string `json:"cards"` +} + type CreateAlertRuleReq struct { CLusterId string `json:"clusterId"` ClusterName string `json:"clusterName"` From eb9d633b17d2c578c2881d56a15d36535a9f961e Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Thu, 9 May 2024 19:33:22 +0800 Subject: [PATCH 13/20] algorithm logic code update Former-commit-id: 816a4270bcf745710acecec7854e1c9331f55556 --- .../schedule/downloadalgothmcodelogic.go | 30 +++++++++++++++++++ .../schedule/getcomputecardsbyclusterlogic.go | 30 +++++++++++++++++++ .../logic/schedule/uploadalgothmcodelogic.go | 30 +++++++++++++++++++ 3 files changed, 90 insertions(+) create mode 100644 api/internal/logic/schedule/downloadalgothmcodelogic.go create mode 100644 api/internal/logic/schedule/getcomputecardsbyclusterlogic.go create mode 100644 api/internal/logic/schedule/uploadalgothmcodelogic.go diff --git a/api/internal/logic/schedule/downloadalgothmcodelogic.go b/api/internal/logic/schedule/downloadalgothmcodelogic.go new file mode 100644 index 00000000..e39f7651 --- /dev/null +++ b/api/internal/logic/schedule/downloadalgothmcodelogic.go @@ -0,0 +1,30 @@ +package schedule + +import ( + "context" + + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + + "github.com/zeromicro/go-zero/core/logx" +) + +type DownloadAlgothmCodeLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext +} + +func NewDownloadAlgothmCodeLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DownloadAlgothmCodeLogic { + return &DownloadAlgothmCodeLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + } +} + +func (l *DownloadAlgothmCodeLogic) DownloadAlgothmCode(req *types.DownloadAlgorithmCodeReq) (resp *types.DownloadAlgorithmCodeResp, err error) { + // todo: add your logic here and delete this line + + return +} diff --git a/api/internal/logic/schedule/getcomputecardsbyclusterlogic.go b/api/internal/logic/schedule/getcomputecardsbyclusterlogic.go new file mode 100644 index 00000000..772a5ce6 --- /dev/null +++ b/api/internal/logic/schedule/getcomputecardsbyclusterlogic.go @@ -0,0 +1,30 @@ +package schedule + +import ( + "context" + + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + + "github.com/zeromicro/go-zero/core/logx" +) + +type GetComputeCardsByClusterLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext +} + +func NewGetComputeCardsByClusterLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetComputeCardsByClusterLogic { + return &GetComputeCardsByClusterLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + } +} + +func (l *GetComputeCardsByClusterLogic) GetComputeCardsByCluster(req *types.GetComputeCardsByClusterReq) (resp *types.GetComputeCardsByClusterResp, err error) { + // todo: add your logic here and delete this line + + return +} diff --git a/api/internal/logic/schedule/uploadalgothmcodelogic.go b/api/internal/logic/schedule/uploadalgothmcodelogic.go new file mode 100644 index 00000000..a0771b04 --- /dev/null +++ b/api/internal/logic/schedule/uploadalgothmcodelogic.go @@ -0,0 +1,30 @@ +package schedule + +import ( + "context" + + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + + "github.com/zeromicro/go-zero/core/logx" +) + +type UploadAlgothmCodeLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext +} + +func NewUploadAlgothmCodeLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UploadAlgothmCodeLogic { + return &UploadAlgothmCodeLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + } +} + +func (l *UploadAlgothmCodeLogic) UploadAlgothmCode(req *types.UploadAlgorithmCodeReq) (resp *types.UploadAlgorithmCodeResp, err error) { + // todo: add your logic here and delete this line + + return +} From a10714ed57cbc42b8a0b7e775eddbefbbfbcbee8 Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Fri, 10 May 2024 10:53:24 +0800 Subject: [PATCH 14/20] core task status sync Former-commit-id: 51279892412a0a6c0d835bc33d57313ba7ada0eb --- api/client/types.go | 14 ++ api/internal/logic/core/pushtaskinfologic.go | 42 +++-- api/internal/types/types.go | 178 ------------------- 3 files changed, 44 insertions(+), 190 deletions(-) diff --git a/api/client/types.go b/api/client/types.go index 940c88df..df8db874 100644 --- a/api/client/types.go +++ b/api/client/types.go @@ -5,6 +5,20 @@ import ( "time" ) +var HpcStatusMapping = map[string][]string{ + "Running": {"RUNNING", "RUNNING", "CONFIGURING", "COMPLETING"}, + "Succeeded": {"COMPLETED"}, + "Failed": {"FAILED", "TIMEOUT", "DEADLINE", "OUT_OF_MEMORY", "BOOT_FAIL", "CANCELLED"}, +} + +var AiStatusMapping = map[string]string{ + "PENDING": "Running", +} + +var CloudStatusMapping = map[string]string{ + "PENDING": "Running", +} + type PullTaskInfoReq struct { AdapterId int64 `form:"adapterId"` } diff --git a/api/internal/logic/core/pushtaskinfologic.go b/api/internal/logic/core/pushtaskinfologic.go index 664a8df9..d9efac1b 100644 --- a/api/internal/logic/core/pushtaskinfologic.go +++ b/api/internal/logic/core/pushtaskinfologic.go @@ -73,20 +73,34 @@ func syncTask(gorm *gorm.DB, taskId int64) { if tx.Error != nil { logx.Error(tx.Error) } - // 子状态统一则修改主任务状态 - statusArray := strings.Split(allStatus, ",") - if len(removeRepeatedElement(statusArray)) == 1 { - updateTask(gorm, taskId, statusArray[0]) - } - // 子任务包含失败状态 主任务则失败 - if strings.Contains(allStatus, constants.Failed) { - updateTask(gorm, taskId, constants.Failed) + for pcmStatus, hpcStatus := range clientCore.HpcStatusMapping { + for _, status := range hpcStatus { + // if Failed type status appears in subTask then update mainTask to Failed + if pcmStatus == "Failed" && strings.Contains(allStatus, status) { + updateTask(gorm, taskId, constants.Failed) + return + // no Failed type status in subTask,if Saved type status appears in subTask then update mainTask to Saved + } else if pcmStatus == "Saved" { + if strings.Contains(allStatus, status) { + updateTask(gorm, taskId, constants.Saved) + return + } + // no Failed and Saved type status in subTask,if Running type status appears in subTask then update mainTask to Running + } else if pcmStatus == "Running" { + if strings.Contains(allStatus, status) { + updateTask(gorm, taskId, constants.Running) + return + } + // at last, mainTask should be succeeded + } else { + if strings.Contains(allStatus, status) { + updateTask(gorm, taskId, constants.Succeeded) + return + } + } + } } - if strings.Contains(allStatus, constants.Running) { - updateTask(gorm, taskId, constants.Running) - } - } func updateTask(gorm *gorm.DB, taskId int64, status string) { @@ -98,8 +112,12 @@ func updateTask(gorm *gorm.DB, taskId int64, status string) { if status == constants.Running { task.StartTime = &now } + if task.Status == constants.Failed || task.Status == constants.Succeeded { + task.EndTime = &now + } gorm.Updates(&task) } + } func removeRepeatedElement(arr []string) (newArr []string) { diff --git a/api/internal/types/types.go b/api/internal/types/types.go index 4520f873..f9693ad1 100644 --- a/api/internal/types/types.go +++ b/api/internal/types/types.go @@ -938,184 +938,6 @@ type ListResult struct { List interface{} `json:"list,omitempty"` } -type HpcInfo struct { - Id int64 `json:"id"` // id - TaskId int64 `json:"task_id"` // 任务id - JobId string `json:"job_id"` // 作业id(在第三方系统中的作业id) - AdapterId int64 `json:"adapter_id"` // 执行任务的适配器id - ClusterId int64 `json:"cluster_id"` // 执行任务的集群id - ClusterType string `json:"cluster_type"` // 执行任务的集群类型 - Name string `json:"name"` // 名称 - Status string `json:"status"` // 状态 - CmdScript string `json:"cmd_script"` - StartTime string `json:"start_time"` // 开始时间 - RunningTime int64 `json:"running_time"` // 运行时间 - DerivedEs string `json:"derived_es"` - Cluster string `json:"cluster"` - BlockId int64 `json:"block_id"` - AllocNodes int64 `json:"alloc_nodes"` - AllocCpu int64 `json:"alloc_cpu"` - CardCount int64 `json:"card_count"` // 卡数 - Version string `json:"version"` - Account string `json:"account"` - WorkDir string `json:"work_dir"` // 工作路径 - AssocId int64 `json:"assoc_id"` - ExitCode int64 `json:"exit_code"` - WallTime string `json:"wall_time"` // 最大运行时间 - Result string `json:"result"` // 运行结果 - DeletedAt string `json:"deleted_at"` // 删除时间 - YamlString string `json:"yaml_string"` - AppType string `json:"app_type"` // 应用类型 - AppName string `json:"app_name"` // 应用名称 - Queue string `json:"queue"` // 队列名称 - SubmitType string `json:"submit_type"` // cmd（命令行模式） - NNode string `json:"n_node"` // 节点个数（当指定该参数时，GAP_NODE_STRING必须为""） - StdOutFile string `json:"std_out_file"` // 工作路径/std.err.%j - StdErrFile string `json:"std_err_file"` // 工作路径/std.err.%j - StdInput string `json:"std_input"` - Environment string `json:"environment"` - DeletedFlag int64 `json:"deleted_flag"` // 是否删除（0-否，1-是） - CreatedBy int64 `json:"created_by"` // 创建人 - CreateTime string `json:"created_time"` // 创建时间 - UpdatedBy int64 `json:"updated_by"` // 更新人 - UpdateTime string `json:"updated_time"` // 更新时间 -} - -type CloudInfo struct { - Participant int64 `json:"participant,omitempty"` - Id int64 `json:"id,omitempty"` - TaskId int64 `json:"taskId,omitempty"` - ApiVersion string `json:"apiVersion,omitempty"` - Kind string `json:"kind,omitempty"` - Namespace string `json:"namespace,omitempty"` - Name string `json:"name,omitempty"` - Status string `json:"status,omitempty"` - StartTime string `json:"startTime,omitempty"` - RunningTime int64 `json:"runningTime,omitempty"` - Result string `json:"result,omitempty"` - YamlString string `json:"yamlString,omitempty"` -} - -type AiInfo struct { - ParticipantId int64 `json:"participantId,omitempty"` - TaskId int64 `json:"taskId,omitempty"` - ProjectId string `json:"project_id,omitempty"` - Name string `json:"name,omitempty"` - Status string `json:"status,omitempty"` - StartTime string `json:"startTime,omitempty"` - RunningTime int64 `json:"runningTime,omitempty"` - Result string `json:"result,omitempty"` - JobId string `json:"jobId,omitempty"` - CreateTime string `json:"createTime,omitempty"` - ImageUrl string `json:"imageUrl,omitempty"` - Command string `json:"command,omitempty"` - FlavorId string `json:"flavorId,omitempty"` - SubscriptionId string `json:"subscriptionId,omitempty"` - ItemVersionId string `json:"itemVersionId,omitempty"` -} - -type VmInfo struct { - ParticipantId int64 `json:"participantId,omitempty"` - TaskId int64 `json:"taskId,omitempty"` - Name string `json:"name,omitempty"` - FlavorRef string `json:"flavor_ref,omitempty"` - ImageRef string `json:"image_ref,omitempty"` - NetworkUuid string `json:"network_uuid,omitempty"` - BlockUuid string `json:"block_uuid,omitempty"` - SourceType string `json:"source_type,omitempty"` - DeleteOnTermination bool `json:"delete_on_termination,omitempty"` - Status string `json:"status,omitempty"` - MinCount string `json:"min_count,omitempty"` - Platform string `json:"platform,omitempty"` - Uuid string `json:"uuid,omitempty"` -} - -type PullTaskInfoReq struct { - AdapterId int64 `form:"adapterId"` -} - -type PullTaskInfoResp struct { - HpcInfoList []*HpcInfo `json:"HpcInfoList,omitempty"` - CloudInfoList []*CloudInfo `json:"CloudInfoList,omitempty"` - AiInfoList []*AiInfo `json:"AiInfoList,omitempty"` - VmInfoList []*VmInfo `json:"VmInfoList,omitempty"` -} - -type PushTaskInfoReq struct { - AdapterId int64 `json:"adapterId"` - HpcInfoList []*HpcInfo `json:"hpcInfoList"` - CloudInfoList []*CloudInfo `json:"cloudInfoList"` - AiInfoList []*AiInfo `json:"aiInfoList"` - VmInfoList []*VmInfo `json:"vmInfoList"` -} - -type PushTaskInfoResp struct { - Code int64 `json:"code"` - Msg string `json:"msg"` -} - -type PushResourceInfoReq struct { - AdapterId int64 `json:"adapterId"` - ResourceStats []ResourceStats `json:"resourceStats"` -} - -type PushResourceInfoResp struct { - Code int64 `json:"code"` - Msg string `json:"msg"` -} - -type NoticeInfo struct { - AdapterId int64 `json:"adapterId"` - AdapterName string `json:"adapterName"` - ClusterId int64 `json:"clusterId"` - ClusterName string `json:"clusterName"` - NoticeType string `json:"noticeType"` - TaskName string `json:"taskName"` - Incident string `json:"incident"` -} - -type ListNoticeReq struct { -} - -type ListNoticeResp struct { - Code int64 `json:"code"` - Msg string `json:"msg"` - Data []NoticeInfo `json:"data"` -} - -type PushNoticeReq struct { - NoticeInfo NoticeInfo `json:"noticeInfo"` -} - -type PushNoticeResp struct { - Code int64 `json:"code"` - Msg string `json:"msg"` -} - -type ResourceStats struct { - ClusterId int64 `json:"clusterId"` - Name string `json:"name"` - CpuCoreAvail int64 `json:"cpuCoreAvail"` - CpuCoreTotal int64 `json:"cpuCoreTotal"` - MemAvail float64 `json:"memAvail"` - MemTotal float64 `json:"memTotal"` - DiskAvail float64 `json:"diskAvail"` - DiskTotal float64 `json:"diskTotal"` - GpuAvail int64 `json:"gpuAvail"` - CardsAvail []*Card `json:"cardsAvail"` - CpuCoreHours float64 `json:"cpuCoreHours"` - Balance float64 `json:"balance"` -} - -type Card struct { - Platform string `json:"platform"` - Type string `json:"type"` - Name string `json:"name"` - TOpsAtFp16 float64 `json:"TOpsAtFp16"` - CardHours float64 `json:"cardHours"` - CardNum int32 `json:"cardNum"` -} - type TaskStatusResp struct { Succeeded int `json:"Succeeded"` Failed int `json:"Failed"` From 275e14a75849b8a70b8b1a16807cc338ecdc8641 Mon Sep 17 00:00:00 2001 From: tzwang Date: Fri, 10 May 2024 16:17:33 +0800 Subject: [PATCH 15/20] added updating aitask status to taskList Former-commit-id: 8908fee9e07f5d334a6a7fd440d1d62246efa972 --- api/internal/logic/core/tasklistlogic.go | 67 +++++++++++++++++++++++- 1 file changed, 66 insertions(+), 1 deletion(-) diff --git a/api/internal/logic/core/tasklistlogic.go b/api/internal/logic/core/tasklistlogic.go index e56be650..382db2e2 100644 --- a/api/internal/logic/core/tasklistlogic.go +++ b/api/internal/logic/core/tasklistlogic.go @@ -17,6 +17,7 @@ package core import ( "context" "fmt" + "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models" "strconv" "time" @@ -55,6 +56,11 @@ func (l *TaskListLogic) TaskList(req *types.TaskListReq) (resp *types.TaskListRe if len(tasks) == 0 { return nil, nil } + + // 更新智算任务状态 + var ch = make(chan struct{}) + go l.updateAitaskStatus(tasks, ch) + // 查询任务总数 l.svcCtx.DbEngin.Model(&models.Task{}).Count(&resp.TotalCount) @@ -106,5 +112,64 @@ func (l *TaskListLogic) TaskList(req *types.TaskListReq) (resp *types.TaskListRe } - return + select { + case _ = <-ch: + return resp, nil + case <-time.After(1 * time.Second): + return resp, nil + } +} + +func (l *TaskListLogic) updateAitaskStatus(tasks []models.Task, ch chan<- struct{}) { + for _, task := range tasks { + if task.AdapterTypeDict != 1 { + continue + } + if task.Status == constants.Succeeded { + continue + } + + var aiTask []*models.TaskAi + tx := l.svcCtx.DbEngin.Raw("select * from task_ai where `task_id` = ? ", task.Id).Scan(&aiTask) + if tx.Error != nil { + logx.Errorf(tx.Error.Error()) + return + } + + start, _ := time.ParseInLocation(constants.Layout, aiTask[0].StartTime, time.Local) + end, _ := time.ParseInLocation(constants.Layout, aiTask[0].EndTime, time.Local) + var status = constants.Succeeded + for _, a := range aiTask { + s, _ := time.ParseInLocation(constants.Layout, a.StartTime, time.Local) + e, _ := time.ParseInLocation(constants.Layout, a.EndTime, time.Local) + + if s.Before(start) { + start = s + } + + if e.After(end) { + end = e + } + + if a.Status == constants.Failed { + status = a.Status + break + } + + if a.Status == constants.Running { + status = a.Status + continue + } + } + + task.Status = status + task.StartTime = &start + task.EndTime = &end + + tx = l.svcCtx.DbEngin.Updates(task) + if tx.Error != nil { + return + } + } + ch <- struct{}{} } From 3c4e18f8d470491b95ac1b20f3420a73f872f915 Mon Sep 17 00:00:00 2001 From: tzwang Date: Fri, 10 May 2024 17:20:52 +0800 Subject: [PATCH 16/20] updated api desc Former-commit-id: 1fcf13a7f363e7e200f560873e3faa62826bce8d --- api/desc/pcm.api | 4 ++-- api/desc/schedule/pcm-schedule.api | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/api/desc/pcm.api b/api/desc/pcm.api index 73c6577d..9eb14b17 100644 --- a/api/desc/pcm.api +++ b/api/desc/pcm.api @@ -968,10 +968,10 @@ service pcm { post /schedule/getOverview returns (ScheduleOverviewResp) @handler DownloadAlgothmCodeHandler - get /schedule/getDownloadAlgothmCode (DownloadAlgorithmCodeReq) returns (DownloadAlgorithmCodeResp) + get /schedule/downloadAlgorithmCodeReq (DownloadAlgorithmCodeReq) returns (DownloadAlgorithmCodeResp) @handler UploadAlgothmCodeHandler - post /schedule/getDownloadAlgothmCode (UploadAlgorithmCodeReq) returns (UploadAlgorithmCodeResp) + post /schedule/uploadAlgorithmCode (UploadAlgorithmCodeReq) returns (UploadAlgorithmCodeResp) @handler GetComputeCardsByClusterHandler get /schedule/getComputeCardsByCluster/:adapterId/:clusterId (GetComputeCardsByClusterReq) returns (GetComputeCardsByClusterResp) diff --git a/api/desc/schedule/pcm-schedule.api b/api/desc/schedule/pcm-schedule.api index 776851db..99574110 100644 --- a/api/desc/schedule/pcm-schedule.api +++ b/api/desc/schedule/pcm-schedule.api @@ -109,7 +109,6 @@ type ( TaskType string `form:"taskType"` Dataset string `form:"dataset"` Algorithm string `form:"algorithm"` - Code string `form:"code"` } DownloadAlgorithmCodeResp { From c5def7491fe7b79b116f6ea3ff5c1c0b5405fdbd Mon Sep 17 00:00:00 2001 From: jagger Date: Fri, 10 May 2024 17:28:02 +0800 Subject: [PATCH 17/20] fix Signed-off-by: jagger Former-commit-id: 5e50a3782fbd4227fc7d107f34f2a7cc570b3f7a --- api/desc/core/pcm-core.api | 12 ++ api/desc/pcm.api | 3 + api/internal/handler/cloud/podlogshandler.go | 24 +++ api/internal/handler/routes.go | 5 + .../logic/cloud/commitgeneraltasklogic.go | 5 +- api/internal/logic/cloud/podlogslogic.go | 30 +++ api/internal/logic/core/pagelisttasklogic.go | 15 +- api/internal/types/types.go | 189 ++++++++++++++++++ pkg/models/cloud/task_cloud.go | 1 + 9 files changed, 274 insertions(+), 10 deletions(-) create mode 100644 api/internal/handler/cloud/podlogshandler.go create mode 100644 api/internal/logic/cloud/podlogslogic.go diff --git a/api/desc/core/pcm-core.api b/api/desc/core/pcm-core.api index f433416e..ccc57605 100644 --- a/api/desc/core/pcm-core.api +++ b/api/desc/core/pcm-core.api @@ -166,6 +166,18 @@ type ( ReqBody []string `json:"reqBody"` Replicas int64 `json:"replicas,string"` } + + PodLogsReq { + TaskId string `json:"taskId"` + TaskName string `json:"taskName"` + ClusterId string `json:"clusterId"` + ClusterName string `json:"clusterName"` + AdapterId string `json:"adapterId"` + AdapterName string `json:"adapterName"` + PodName string `json:"podName"` + stream bool `json:"stream"` + + } ) type deleteTaskReq { diff --git a/api/desc/pcm.api b/api/desc/pcm.api index 73c6577d..e6fd637d 100644 --- a/api/desc/pcm.api +++ b/api/desc/pcm.api @@ -231,6 +231,9 @@ service pcm { @doc "Create cloud computing common tasks" @handler commitGeneralTask post /cloud/task/create (GeneralTaskReq) returns () + + @handler podLogs + post /cloud/pod/logs (PodLogsReq) returns (string) } //智算二级接口 diff --git a/api/internal/handler/cloud/podlogshandler.go b/api/internal/handler/cloud/podlogshandler.go new file mode 100644 index 00000000..78824344 --- /dev/null +++ b/api/internal/handler/cloud/podlogshandler.go @@ -0,0 +1,24 @@ +package cloud + +import ( + "github.com/zeromicro/go-zero/rest/httpx" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/cloud" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" + "net/http" +) + +func PodLogsHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req types.PodLogsReq + if err := httpx.Parse(r, &req); err != nil { + result.ParamErrorResult(r, w, err) + return + } + + l := cloud.NewPodLogsLogic(r.Context(), svcCtx, w) + resp, err := l.PodLogs(&req, w) + result.HttpResult(r, w, resp, err) + } +} diff --git a/api/internal/handler/routes.go b/api/internal/handler/routes.go index 0a2e86a0..918c8057 100644 --- a/api/internal/handler/routes.go +++ b/api/internal/handler/routes.go @@ -277,6 +277,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { Path: "/cloud/task/create", Handler: cloud.CommitGeneralTaskHandler(serverCtx), }, + { + Method: http.MethodPost, + Path: "/cloud/pod/logs", + Handler: cloud.PodLogsHandler(serverCtx), + }, }, rest.WithPrefix("/pcm/v1"), ) diff --git a/api/internal/logic/cloud/commitgeneraltasklogic.go b/api/internal/logic/cloud/commitgeneraltasklogic.go index 0162b832..14240bc0 100644 --- a/api/internal/logic/cloud/commitgeneraltasklogic.go +++ b/api/internal/logic/cloud/commitgeneraltasklogic.go @@ -99,6 +99,8 @@ func (l *CommitGeneralTaskLogic) CommitGeneralTask(req *types.GeneralTaskReq) er Strategy: strategy, } var taskClouds []cloud.TaskCloudModel + adapterName := "" + tx.Table("t_adapter").Select("name").Where("id=?", adapterId).Find(&adapterName) for _, r := range rs { for _, s := range req.ReqBody { sStruct := UnMarshalK8sStruct(s, int64(r.Replica)) @@ -107,6 +109,7 @@ func (l *CommitGeneralTaskLogic) CommitGeneralTask(req *types.GeneralTaskReq) er taskCloud.TaskId = uint(taskModel.Id) clusterId, _ := strconv.ParseUint(r.ClusterId, 10, 64) taskCloud.AdapterId = uint(adapterId) + taskCloud.AdapterName = adapterName taskCloud.ClusterId = uint(clusterId) taskCloud.ClusterName = r.ClusterName taskCloud.Status = constants.Saved @@ -116,8 +119,6 @@ func (l *CommitGeneralTaskLogic) CommitGeneralTask(req *types.GeneralTaskReq) er taskClouds = append(taskClouds, taskCloud) } } - adapterName := "" - tx.Table("t_adapter").Select("name").Where("id=?", adapterId).Find(&adapterName) noticeInfo := clientCore.NoticeInfo{ AdapterId: int64(adapterId), AdapterName: adapterName, diff --git a/api/internal/logic/cloud/podlogslogic.go b/api/internal/logic/cloud/podlogslogic.go new file mode 100644 index 00000000..480c2d21 --- /dev/null +++ b/api/internal/logic/cloud/podlogslogic.go @@ -0,0 +1,30 @@ +package cloud + +import ( + "context" + "github.com/zeromicro/go-zero/core/logx" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + "net/http" +) + +type PodLogsLogic struct { + logx.Logger + ctx context.Context + svcCtx *svc.ServiceContext + w http.ResponseWriter +} + +func NewPodLogsLogic(ctx context.Context, svcCtx *svc.ServiceContext, w http.ResponseWriter) *PodLogsLogic { + return &PodLogsLogic{ + Logger: logx.WithContext(ctx), + ctx: ctx, + svcCtx: svcCtx, + w: w, + } +} + +func (l *PodLogsLogic) PodLogs(req *types.PodLogsReq, w http.ResponseWriter) (resp string, err error) { + // todo: add your logic here and delete this line + return +} diff --git a/api/internal/logic/core/pagelisttasklogic.go b/api/internal/logic/core/pagelisttasklogic.go index ce8ae65d..be3baccd 100644 --- a/api/internal/logic/core/pagelisttasklogic.go +++ b/api/internal/logic/core/pagelisttasklogic.go @@ -2,13 +2,12 @@ package core import ( "context" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" + "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils/timeutils" "time" - "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" - "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" - "github.com/zeromicro/go-zero/core/logx" ) @@ -51,15 +50,15 @@ func (l *PageListTaskLogic) PageListTask(req *types.PageTaskReq) (resp *types.Pa return nil, result.NewDefaultError(err.Error()) } for _, model := range list { - if model.EndTime != "" && model.StartTime != "" { + if model.StartTime != "" && model.EndTime == "" { + startTime := timeutils.TimeStringToGoTime(model.StartTime) + model.RunningTime = int64(time.Now().Sub(startTime).Seconds()) + } + if model.StartTime != "" && model.EndTime != "" { startTime := timeutils.TimeStringToGoTime(model.StartTime) endTime := timeutils.TimeStringToGoTime(model.EndTime) model.RunningTime = int64(endTime.Sub(startTime).Seconds()) } - if model.StartTime != "" { - startTime := timeutils.TimeStringToGoTime(model.StartTime) - model.RunningTime = int64(time.Now().Sub(startTime).Seconds()) - } } resp.List = &list resp.PageSize = req.PageSize diff --git a/api/internal/types/types.go b/api/internal/types/types.go index f9693ad1..2e9b99ac 100644 --- a/api/internal/types/types.go +++ b/api/internal/types/types.go @@ -150,6 +150,17 @@ type GeneralTaskReq struct { Replicas int64 `json:"replicas,string"` } +type PodLogsReq struct { + TaskId string `json:"taskId"` + TaskName string `json:"taskName"` + ClusterId string `json:"clusterId"` + ClusterName string `json:"clusterName"` + AdapterId string `json:"adapterId"` + AdapterName string `json:"adapterName"` + PodName string `json:"podName"` + Stream bool `json:"stream"` +} + type DeleteTaskReq struct { Id int64 `path:"id"` } @@ -938,6 +949,184 @@ type ListResult struct { List interface{} `json:"list,omitempty"` } +type HpcInfo struct { + Id int64 `json:"id"` // id + TaskId int64 `json:"task_id"` // 任务id + JobId string `json:"job_id"` // 作业id(在第三方系统中的作业id) + AdapterId int64 `json:"adapter_id"` // 执行任务的适配器id + ClusterId int64 `json:"cluster_id"` // 执行任务的集群id + ClusterType string `json:"cluster_type"` // 执行任务的集群类型 + Name string `json:"name"` // 名称 + Status string `json:"status"` // 状态 + CmdScript string `json:"cmd_script"` + StartTime string `json:"start_time"` // 开始时间 + RunningTime int64 `json:"running_time"` // 运行时间 + DerivedEs string `json:"derived_es"` + Cluster string `json:"cluster"` + BlockId int64 `json:"block_id"` + AllocNodes int64 `json:"alloc_nodes"` + AllocCpu int64 `json:"alloc_cpu"` + CardCount int64 `json:"card_count"` // 卡数 + Version string `json:"version"` + Account string `json:"account"` + WorkDir string `json:"work_dir"` // 工作路径 + AssocId int64 `json:"assoc_id"` + ExitCode int64 `json:"exit_code"` + WallTime string `json:"wall_time"` // 最大运行时间 + Result string `json:"result"` // 运行结果 + DeletedAt string `json:"deleted_at"` // 删除时间 + YamlString string `json:"yaml_string"` + AppType string `json:"app_type"` // 应用类型 + AppName string `json:"app_name"` // 应用名称 + Queue string `json:"queue"` // 队列名称 + SubmitType string `json:"submit_type"` // cmd（命令行模式） + NNode string `json:"n_node"` // 节点个数（当指定该参数时，GAP_NODE_STRING必须为""） + StdOutFile string `json:"std_out_file"` // 工作路径/std.err.%j + StdErrFile string `json:"std_err_file"` // 工作路径/std.err.%j + StdInput string `json:"std_input"` + Environment string `json:"environment"` + DeletedFlag int64 `json:"deleted_flag"` // 是否删除（0-否，1-是） + CreatedBy int64 `json:"created_by"` // 创建人 + CreateTime string `json:"created_time"` // 创建时间 + UpdatedBy int64 `json:"updated_by"` // 更新人 + UpdateTime string `json:"updated_time"` // 更新时间 +} + +type CloudInfo struct { + Participant int64 `json:"participant,omitempty"` + Id int64 `json:"id,omitempty"` + TaskId int64 `json:"taskId,omitempty"` + ApiVersion string `json:"apiVersion,omitempty"` + Kind string `json:"kind,omitempty"` + Namespace string `json:"namespace,omitempty"` + Name string `json:"name,omitempty"` + Status string `json:"status,omitempty"` + StartTime string `json:"startTime,omitempty"` + RunningTime int64 `json:"runningTime,omitempty"` + Result string `json:"result,omitempty"` + YamlString string `json:"yamlString,omitempty"` +} + +type AiInfo struct { + ParticipantId int64 `json:"participantId,omitempty"` + TaskId int64 `json:"taskId,omitempty"` + ProjectId string `json:"project_id,omitempty"` + Name string `json:"name,omitempty"` + Status string `json:"status,omitempty"` + StartTime string `json:"startTime,omitempty"` + RunningTime int64 `json:"runningTime,omitempty"` + Result string `json:"result,omitempty"` + JobId string `json:"jobId,omitempty"` + CreateTime string `json:"createTime,omitempty"` + ImageUrl string `json:"imageUrl,omitempty"` + Command string `json:"command,omitempty"` + FlavorId string `json:"flavorId,omitempty"` + SubscriptionId string `json:"subscriptionId,omitempty"` + ItemVersionId string `json:"itemVersionId,omitempty"` +} + +type VmInfo struct { + ParticipantId int64 `json:"participantId,omitempty"` + TaskId int64 `json:"taskId,omitempty"` + Name string `json:"name,omitempty"` + FlavorRef string `json:"flavor_ref,omitempty"` + ImageRef string `json:"image_ref,omitempty"` + NetworkUuid string `json:"network_uuid,omitempty"` + BlockUuid string `json:"block_uuid,omitempty"` + SourceType string `json:"source_type,omitempty"` + DeleteOnTermination bool `json:"delete_on_termination,omitempty"` + Status string `json:"status,omitempty"` + MinCount string `json:"min_count,omitempty"` + Platform string `json:"platform,omitempty"` + Uuid string `json:"uuid,omitempty"` +} + +type PullTaskInfoReq struct { + AdapterId int64 `form:"adapterId"` +} + +type PullTaskInfoResp struct { + HpcInfoList []*HpcInfo `json:"HpcInfoList,omitempty"` + CloudInfoList []*CloudInfo `json:"CloudInfoList,omitempty"` + AiInfoList []*AiInfo `json:"AiInfoList,omitempty"` + VmInfoList []*VmInfo `json:"VmInfoList,omitempty"` +} + +type PushTaskInfoReq struct { + AdapterId int64 `json:"adapterId"` + HpcInfoList []*HpcInfo `json:"hpcInfoList"` + CloudInfoList []*CloudInfo `json:"cloudInfoList"` + AiInfoList []*AiInfo `json:"aiInfoList"` + VmInfoList []*VmInfo `json:"vmInfoList"` +} + +type PushTaskInfoResp struct { + Code int64 `json:"code"` + Msg string `json:"msg"` +} + +type PushResourceInfoReq struct { + AdapterId int64 `json:"adapterId"` + ResourceStats []ResourceStats `json:"resourceStats"` +} + +type PushResourceInfoResp struct { + Code int64 `json:"code"` + Msg string `json:"msg"` +} + +type NoticeInfo struct { + AdapterId int64 `json:"adapterId"` + AdapterName string `json:"adapterName"` + ClusterId int64 `json:"clusterId"` + ClusterName string `json:"clusterName"` + NoticeType string `json:"noticeType"` + TaskName string `json:"taskName"` + Incident string `json:"incident"` +} + +type ListNoticeReq struct { +} + +type ListNoticeResp struct { + Code int64 `json:"code"` + Msg string `json:"msg"` + Data []NoticeInfo `json:"data"` +} + +type PushNoticeReq struct { + NoticeInfo NoticeInfo `json:"noticeInfo"` +} + +type PushNoticeResp struct { + Code int64 `json:"code"` + Msg string `json:"msg"` +} + +type ResourceStats struct { + ClusterId int64 `json:"clusterId"` + Name string `json:"name"` + CpuCoreAvail int64 `json:"cpuCoreAvail"` + CpuCoreTotal int64 `json:"cpuCoreTotal"` + MemAvail float64 `json:"memAvail"` + MemTotal float64 `json:"memTotal"` + DiskAvail float64 `json:"diskAvail"` + DiskTotal float64 `json:"diskTotal"` + GpuAvail int64 `json:"gpuAvail"` + CardsAvail []*Card `json:"cardsAvail"` + CpuCoreHours float64 `json:"cpuCoreHours"` + Balance float64 `json:"balance"` +} + +type Card struct { + Platform string `json:"platform"` + Type string `json:"type"` + Name string `json:"name"` + TOpsAtFp16 float64 `json:"TOpsAtFp16"` + CardHours float64 `json:"cardHours"` + CardNum int32 `json:"cardNum"` +} + type TaskStatusResp struct { Succeeded int `json:"Succeeded"` Failed int `json:"Failed"` diff --git a/pkg/models/cloud/task_cloud.go b/pkg/models/cloud/task_cloud.go index d60c236e..90949145 100644 --- a/pkg/models/cloud/task_cloud.go +++ b/pkg/models/cloud/task_cloud.go @@ -8,6 +8,7 @@ import ( type TaskCloudModel struct { Id uint `json:"id" gorm:"primarykey;not null;comment:id"` TaskId uint `json:"taskId" gorm:"not null;comment:task表id"` + AdapterName string `json:"adapterName" gorm:"not null;comment:适配器名称"` AdapterId uint `json:"adapterId" gorm:"not null;comment:适配器id"` ClusterId uint `json:"clusterId" gorm:"not null;comment:集群id"` ClusterName string `json:"clusterName" gorm:"not null;comment:集群名称"` From 3392fa01ac6eab46466110ffebe1b580e20a4c1a Mon Sep 17 00:00:00 2001 From: tzwang Date: Fri, 10 May 2024 21:50:03 +0800 Subject: [PATCH 18/20] updated downloadalgorithmcode logic Former-commit-id: 75272c3ecdea08e482f3f78125ec3019a833095e --- api/desc/pcm.api | 2 +- api/desc/schedule/pcm-schedule.api | 2 +- api/internal/handler/routes.go | 4 +- .../schedule/downloadalgothmcodehandler.go | 11 ++--- .../schedule/uploadalgothmcodehandler.go | 11 ++--- .../schedule/downloadalgothmcodelogic.go | 12 ++++-- .../logic/schedule/uploadalgothmcodelogic.go | 11 +++-- api/internal/storeLink/octopus.go | 40 ++++++++++++++++++- api/internal/types/types.go | 3 +- go.mod | 2 +- go.sum | 4 +- 11 files changed, 72 insertions(+), 30 deletions(-) diff --git a/api/desc/pcm.api b/api/desc/pcm.api index 31116667..03779d46 100644 --- a/api/desc/pcm.api +++ b/api/desc/pcm.api @@ -971,7 +971,7 @@ service pcm { post /schedule/getOverview returns (ScheduleOverviewResp) @handler DownloadAlgothmCodeHandler - get /schedule/downloadAlgorithmCodeReq (DownloadAlgorithmCodeReq) returns (DownloadAlgorithmCodeResp) + get /schedule/downloadAlgorithmCode (DownloadAlgorithmCodeReq) returns (DownloadAlgorithmCodeResp) @handler UploadAlgothmCodeHandler post /schedule/uploadAlgorithmCode (UploadAlgorithmCodeReq) returns (UploadAlgorithmCodeResp) diff --git a/api/desc/schedule/pcm-schedule.api b/api/desc/schedule/pcm-schedule.api index 99574110..2fd6e4f8 100644 --- a/api/desc/schedule/pcm-schedule.api +++ b/api/desc/schedule/pcm-schedule.api @@ -112,7 +112,7 @@ type ( } DownloadAlgorithmCodeResp { - Code string `json:"algorithms"` + Code string `json:"code"` } UploadAlgorithmCodeReq { diff --git a/api/internal/handler/routes.go b/api/internal/handler/routes.go index 918c8057..d63d02e1 100644 --- a/api/internal/handler/routes.go +++ b/api/internal/handler/routes.go @@ -1212,12 +1212,12 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) { }, { Method: http.MethodGet, - Path: "/schedule/getDownloadAlgothmCode", + Path: "/schedule/downloadAlgorithmCode", Handler: schedule.DownloadAlgothmCodeHandler(serverCtx), }, { Method: http.MethodPost, - Path: "/schedule/getDownloadAlgothmCode", + Path: "/schedule/uploadAlgorithmCode", Handler: schedule.UploadAlgothmCodeHandler(serverCtx), }, { diff --git a/api/internal/handler/schedule/downloadalgothmcodehandler.go b/api/internal/handler/schedule/downloadalgothmcodehandler.go index 14207bba..e70d85ea 100644 --- a/api/internal/handler/schedule/downloadalgothmcodehandler.go +++ b/api/internal/handler/schedule/downloadalgothmcodehandler.go @@ -7,22 +7,19 @@ import ( "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/schedule" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" ) func DownloadAlgothmCodeHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req types.DownloadAlgorithmCodeReq if err := httpx.Parse(r, &req); err != nil { - httpx.ErrorCtx(r.Context(), w, err) + result.ParamErrorResult(r, w, err) return } l := schedule.NewDownloadAlgothmCodeLogic(r.Context(), svcCtx) - resp, err := l.DownloadAlgothmCode(&req) - if err != nil { - httpx.ErrorCtx(r.Context(), w, err) - } else { - httpx.OkJsonCtx(r.Context(), w, resp) - } + resp, err := l.DownloadAlgorithmCode(&req) + result.HttpResult(r, w, resp, err) } } diff --git a/api/internal/handler/schedule/uploadalgothmcodehandler.go b/api/internal/handler/schedule/uploadalgothmcodehandler.go index bfbd05d1..681715e2 100644 --- a/api/internal/handler/schedule/uploadalgothmcodehandler.go +++ b/api/internal/handler/schedule/uploadalgothmcodehandler.go @@ -7,22 +7,19 @@ import ( "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/logic/schedule" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc" "gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types" + "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" ) func UploadAlgothmCodeHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req types.UploadAlgorithmCodeReq if err := httpx.Parse(r, &req); err != nil { - httpx.ErrorCtx(r.Context(), w, err) + result.ParamErrorResult(r, w, err) return } l := schedule.NewUploadAlgothmCodeLogic(r.Context(), svcCtx) - resp, err := l.UploadAlgothmCode(&req) - if err != nil { - httpx.ErrorCtx(r.Context(), w, err) - } else { - httpx.OkJsonCtx(r.Context(), w, resp) - } + resp, err := l.UploadAlgorithmCode(&req) + result.HttpResult(r, w, resp, err) } } diff --git a/api/internal/logic/schedule/downloadalgothmcodelogic.go b/api/internal/logic/schedule/downloadalgothmcodelogic.go index e39f7651..81b96579 100644 --- a/api/internal/logic/schedule/downloadalgothmcodelogic.go +++ b/api/internal/logic/schedule/downloadalgothmcodelogic.go @@ -23,8 +23,14 @@ func NewDownloadAlgothmCodeLogic(ctx context.Context, svcCtx *svc.ServiceContext } } -func (l *DownloadAlgothmCodeLogic) DownloadAlgothmCode(req *types.DownloadAlgorithmCodeReq) (resp *types.DownloadAlgorithmCodeResp, err error) { - // todo: add your logic here and delete this line +func (l *DownloadAlgothmCodeLogic) DownloadAlgorithmCode(req *types.DownloadAlgorithmCodeReq) (resp *types.DownloadAlgorithmCodeResp, err error) { + resp = &types.DownloadAlgorithmCodeResp{} + code, err := l.svcCtx.Scheduler.AiService.AiCollectorAdapterMap[req.AdapterId][req.ClusterId].DownloadAlgorithmCode(l.ctx, + req.ResourceType, req.Card, req.TaskType, req.Dataset, req.Algorithm) + if err != nil { + return nil, err + } + resp.Code = code - return + return resp, nil } diff --git a/api/internal/logic/schedule/uploadalgothmcodelogic.go b/api/internal/logic/schedule/uploadalgothmcodelogic.go index a0771b04..052412c4 100644 --- a/api/internal/logic/schedule/uploadalgothmcodelogic.go +++ b/api/internal/logic/schedule/uploadalgothmcodelogic.go @@ -23,8 +23,13 @@ func NewUploadAlgothmCodeLogic(ctx context.Context, svcCtx *svc.ServiceContext) } } -func (l *UploadAlgothmCodeLogic) UploadAlgothmCode(req *types.UploadAlgorithmCodeReq) (resp *types.UploadAlgorithmCodeResp, err error) { - // todo: add your logic here and delete this line +func (l *UploadAlgothmCodeLogic) UploadAlgorithmCode(req *types.UploadAlgorithmCodeReq) (resp *types.UploadAlgorithmCodeResp, err error) { + resp = &types.UploadAlgorithmCodeResp{} + err = l.svcCtx.Scheduler.AiService.AiCollectorAdapterMap[req.AdapterId][req.ClusterId].UploadAlgorithmCode(l.ctx, + req.ResourceType, req.Card, req.TaskType, req.Dataset, req.Algorithm, req.Code) + if err != nil { + return nil, err + } - return + return resp, nil } diff --git a/api/internal/storeLink/octopus.go b/api/internal/storeLink/octopus.go index 8e1f7af0..a088e56a 100644 --- a/api/internal/storeLink/octopus.go +++ b/api/internal/storeLink/octopus.go @@ -54,6 +54,7 @@ const ( CAMBRICON = "cambricon" TRAIN_CMD = "cd /code; python train.py" VERSION = "V1" + DOMAIN = "http://192.168.242.41:8001/" ) var ( @@ -340,7 +341,44 @@ func (o *OctopusLink) GetAlgorithms(ctx context.Context) ([]*collector.Algorithm } func (o *OctopusLink) DownloadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string) (string, error) { - return "", nil + dcReq := &octopus.DownloadCompressReq{ + Platform: o.platform, + Version: VERSION, + AlgorithmId: "", + } + dcResp, err := o.octopusRpc.DownloadCompress(ctx, dcReq) + if err != nil { + return "", err + } + + if !dcResp.Success { + return "", errors.New(dcResp.Error.Message) + } + + daReq := &octopus.DownloadAlgorithmReq{ + Platform: o.platform, + Version: VERSION, + AlgorithmId: "", + CompressAt: dcResp.Payload.CompressAt, + Domain: DOMAIN, + } + daResp, err := o.octopusRpc.DownloadAlgorithm(ctx, daReq) + if err != nil { + return "", err + } + if !daResp.Success { + return "", errors.New(dcResp.Error.Message) + } + urlReq := &octopus.AlgorithmUrlReq{ + Platform: o.platform, + Url: daResp.Payload.DownloadUrl, + } + urlResp, err := o.octopusRpc.DownloadAlgorithmUrl(ctx, urlReq) + if err != nil { + return "", err + } + + return urlResp.Algorithm, nil } func (o *OctopusLink) UploadAlgorithmCode(ctx context.Context, resourceType string, card string, taskType string, dataset string, algorithm string, code string) error { diff --git a/api/internal/types/types.go b/api/internal/types/types.go index 2e9b99ac..eef38e23 100644 --- a/api/internal/types/types.go +++ b/api/internal/types/types.go @@ -5695,11 +5695,10 @@ type DownloadAlgorithmCodeReq struct { TaskType string `form:"taskType"` Dataset string `form:"dataset"` Algorithm string `form:"algorithm"` - Code string `form:"code"` } type DownloadAlgorithmCodeResp struct { - Code string `json:"algorithms"` + Code string `json:"code"` } type UploadAlgorithmCodeReq struct { diff --git a/go.mod b/go.mod index fb46a881..bf50e2ee 100644 --- a/go.mod +++ b/go.mod @@ -26,7 +26,7 @@ require ( github.com/zeromicro/go-zero v1.6.3 gitlink.org.cn/JointCloud/pcm-ac v0.0.0-20240426095603-549fefd8bece gitlink.org.cn/JointCloud/pcm-kubernetes v0.0.0-20240301071143-347480abff2c - gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240424085753-6899615e9142 + gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240510133934-6a5526289b35 gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203 gitlink.org.cn/JointCloud/pcm-slurm v0.0.0-20240301080743-8b94bbaf57f5 gitlink.org.cn/jcce-pcm/pcm-participant-ceph v0.0.0-20230904090036-24fc730ec87d diff --git a/go.sum b/go.sum index c8224c75..e1d1ec5b 100644 --- a/go.sum +++ b/go.sum @@ -1082,8 +1082,8 @@ gitlink.org.cn/JointCloud/pcm-ac v0.0.0-20240426095603-549fefd8bece h1:W3yBnvAVV gitlink.org.cn/JointCloud/pcm-ac v0.0.0-20240426095603-549fefd8bece/go.mod h1:w3Nb5TNymCItQ7K3x4Q0JLuoq9OerwAzAWT2zsPE9Xo= gitlink.org.cn/JointCloud/pcm-kubernetes v0.0.0-20240301071143-347480abff2c h1:2Wl/hvaSFjh6fmCSIQhjkr9llMRREQeqcXNLZ/HPY18= gitlink.org.cn/JointCloud/pcm-kubernetes v0.0.0-20240301071143-347480abff2c/go.mod h1:lSRfGs+PxFvw7CcndHWRd6UlLlGrZn0b0hp5cfaMNGw= -gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240424085753-6899615e9142 h1:+po0nesBDSWsgCySBG7eEXk7i9Ytd58wqvjL1M9y6d8= -gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240424085753-6899615e9142/go.mod h1:QOD5+/l2D+AYBjF2h5T0mdJyfGAmF78QmeKdbBXbjLQ= +gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240510133934-6a5526289b35 h1:E2QfpS3Y0FjR8Zyv5l2Ti/2NetQFqHG66c8+T/+J1u0= +gitlink.org.cn/JointCloud/pcm-octopus v0.0.0-20240510133934-6a5526289b35/go.mod h1:QOD5+/l2D+AYBjF2h5T0mdJyfGAmF78QmeKdbBXbjLQ= gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203 h1:s6PsZ1+bev294IWdZRlV7mnOwI1+UzFcldVW/BqhQzI= gitlink.org.cn/JointCloud/pcm-openstack v0.0.0-20240403033338-e7edabad4203/go.mod h1:i2rrbMQ+Fve345BY9Heh4MUqVTAimZQElQhzzRee5B8= gitlink.org.cn/JointCloud/pcm-slurm v0.0.0-20240301080743-8b94bbaf57f5 h1:+/5vnzkJBfMRnya1NrhOzlroUtRa5ePiYbPKlHLoLV0= From 0fbbf4c454d4f1632c63385dba7c67b37d6d7ee9 Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Fri, 10 May 2024 22:07:06 +0800 Subject: [PATCH 19/20] send notice in core && adapter and cluster info saved in subtask Former-commit-id: 6d85d7373c62b604c84956d0b82d51670b39b045 --- api/client/types.go | 38 ++--- api/internal/logic/core/pushtaskinfologic.go | 140 +++++++++++++----- api/internal/logic/hpc/commithpctasklogic.go | 19 ++- .../scheduler/schedulers/aiScheduler.go | 8 +- pkg/models/aimodel_gen.go | 29 ++-- pkg/models/cloud/task_cloud.go | 1 + pkg/models/taskhpcmodel_gen.go | 15 +- pkg/models/taskvmmodel_gen.go | 6 +- 8 files changed, 173 insertions(+), 83 deletions(-) diff --git a/api/client/types.go b/api/client/types.go index df8db874..32e22c1b 100644 --- a/api/client/types.go +++ b/api/client/types.go @@ -5,20 +5,12 @@ import ( "time" ) -var HpcStatusMapping = map[string][]string{ +var StatusMapping = map[string][]string{ "Running": {"RUNNING", "RUNNING", "CONFIGURING", "COMPLETING"}, "Succeeded": {"COMPLETED"}, "Failed": {"FAILED", "TIMEOUT", "DEADLINE", "OUT_OF_MEMORY", "BOOT_FAIL", "CANCELLED"}, } -var AiStatusMapping = map[string]string{ - "PENDING": "Running", -} - -var CloudStatusMapping = map[string]string{ - "PENDING": "Running", -} - type PullTaskInfoReq struct { AdapterId int64 `form:"adapterId"` } @@ -58,6 +50,7 @@ type NoticeInfo struct { ClusterId int64 `json:"clusterId"` ClusterName string `json:"clusterName"` NoticeType string `json:"noticeType"` + TaskId int64 `json:"taskId"` TaskName string `json:"taskName"` Incident string `json:"incident"` CreatedTime time.Time `json:"createdTime"` @@ -82,11 +75,13 @@ type PushNoticeResp struct { } type HpcInfo struct { - Id int64 `json:"id"` // id - TaskId int64 `json:"task_id"` // 任务id - JobId string `json:"job_id"` // 作业id(在第三方系统中的作业id) - AdapterId int64 `json:"adapter_id"` // 执行任务的适配器id - ClusterId int64 `json:"cluster_id"` // 执行任务的集群id + Id int64 `json:"id"` // id + TaskId int64 `json:"task_id"` // 任务id + JobId string `json:"job_id"` // 作业id(在第三方系统中的作业id) + AdapterId int64 `json:"adapter_id"` // 执行任务的适配器id + AdapterName string `json:"adapterName,omitempty,optional"` + ClusterId int64 `json:"cluster_id"` // 执行任务的集群id + ClusterName string `json:"clusterName,omitempty,optional"` ClusterType string `json:"cluster_type"` // 执行任务的集群类型 Name string `json:"name"` // 名称 Status string `json:"status"` // 状态 @@ -127,8 +122,9 @@ type HpcInfo struct { type CloudInfo struct { Id uint `json:"id,omitempty,optional"` TaskId int64 `json:"taskId,omitempty,optional"` - AdapterId uint `json:"adapterId,omitempty,optional"` - ClusterId uint `json:"clusterId,omitempty,optional"` + AdapterId int64 `json:"adapterId,omitempty,optional"` + AdapterName string `json:"adapterName,omitempty,optional"` + ClusterId int64 `json:"clusterId,omitempty,optional"` ClusterName string `json:"clusterName,omitempty,optional"` Kind string `json:"kind,omitempty,optional"` Status string `json:"status,omitempty,optional"` @@ -139,9 +135,12 @@ type CloudInfo struct { } type AiInfo struct { - ParticipantId int64 `json:"participantId,omitempty"` TaskId int64 `json:"taskId,omitempty"` ProjectId string `json:"project_id,omitempty"` + AdapterId int64 `json:"adapterId,omitempty,optional"` + AdapterName string `json:"adapterName,omitempty,optional"` + ClusterId int64 `json:"clusterId,omitempty,optional"` + ClusterName string `json:"clusterName,omitempty,optional"` Name string `json:"name,omitempty"` Status string `json:"status,omitempty"` StartTime string `json:"startTime,omitempty"` @@ -157,9 +156,12 @@ type AiInfo struct { } type VmInfo struct { - ParticipantId int64 `json:"participantId,omitempty"` TaskId int64 `json:"taskId,omitempty"` Name string `json:"name,omitempty"` + AdapterId int64 `json:"adapterId,omitempty,optional"` + AdapterName string `json:"adapterName,omitempty,optional"` + ClusterId int64 `json:"clusterId,omitempty,optional"` + ClusterName string `json:"clusterName,omitempty,optional"` FlavorRef string `json:"flavor_ref,omitempty"` ImageRef string `json:"image_ref,omitempty"` NetworkUuid string `json:"network_uuid,omitempty"` diff --git a/api/internal/logic/core/pushtaskinfologic.go b/api/internal/logic/core/pushtaskinfologic.go index d9efac1b..ff5a102f 100644 --- a/api/internal/logic/core/pushtaskinfologic.go +++ b/api/internal/logic/core/pushtaskinfologic.go @@ -41,61 +41,143 @@ func (l *PushTaskInfoLogic) PushTaskInfo(req *clientCore.PushTaskInfoReq) (*clie } l.svcCtx.DbEngin.Exec("update task_cloud set status = ?,start_time = ?,result = ? where task_id = ?", cloudInfo.Status, cloudInfo.StartTime, cloudInfo.Result, cloudInfo.TaskId) - syncTask(l.svcCtx.DbEngin, int64(taskId)) + var taskName string + l.svcCtx.DbEngin.Raw("select name as kind from task where id = ?", taskId).Scan(&taskName) + noticeInfo := clientCore.NoticeInfo{ + TaskId: cloudInfo.TaskId, + AdapterId: cloudInfo.AdapterId, + AdapterName: cloudInfo.AdapterName, + ClusterId: cloudInfo.ClusterId, + ClusterName: cloudInfo.ClusterName, + TaskName: taskName, + } + syncTask(l.svcCtx.DbEngin, noticeInfo) } case 2: for _, hpcInfo := range req.HpcInfoList { l.svcCtx.DbEngin.Exec("update task_hpc set status = ?,start_time = ?,job_id = ? where cluster_id = ? and task_id = ? and name = ?", hpcInfo.Status, hpcInfo.StartTime, hpcInfo.JobId, hpcInfo.ClusterId, hpcInfo.TaskId, hpcInfo.Name) - syncTask(l.svcCtx.DbEngin, hpcInfo.TaskId) + noticeInfo := clientCore.NoticeInfo{ + TaskId: hpcInfo.TaskId, + AdapterId: hpcInfo.AdapterId, + AdapterName: hpcInfo.AdapterName, + ClusterId: hpcInfo.ClusterId, + ClusterName: hpcInfo.ClusterName, + TaskName: hpcInfo.Name, + } + syncTask(l.svcCtx.DbEngin, noticeInfo) } case 1: for _, aiInfo := range req.AiInfoList { l.svcCtx.DbEngin.Exec("update task_ai set status = ?,start_time = ?,project_id = ?,job_id = ? where participant_id = ? and task_id = ? and name = ?", aiInfo.Status, aiInfo.StartTime, aiInfo.ProjectId, aiInfo.JobId, req.AdapterId, aiInfo.TaskId, aiInfo.Name) - syncTask(l.svcCtx.DbEngin, aiInfo.TaskId) + noticeInfo := clientCore.NoticeInfo{ + TaskId: aiInfo.TaskId, + AdapterId: aiInfo.AdapterId, + AdapterName: aiInfo.AdapterName, + ClusterId: aiInfo.ClusterId, + ClusterName: aiInfo.ClusterName, + TaskName: aiInfo.Name, + } + syncTask(l.svcCtx.DbEngin, noticeInfo) } case 3: for _, vmInfo := range req.VmInfoList { l.svcCtx.DbEngin.Exec("update task_vm set status = ?,start_time = ? where participant_id = ? and task_id = ? and name = ?", vmInfo.Status, vmInfo.StartTime, req.AdapterId, vmInfo.TaskId, vmInfo.Name) - syncTask(l.svcCtx.DbEngin, vmInfo.TaskId) + noticeInfo := clientCore.NoticeInfo{ + TaskId: vmInfo.TaskId, + AdapterId: vmInfo.AdapterId, + AdapterName: vmInfo.AdapterName, + ClusterId: vmInfo.ClusterId, + ClusterName: vmInfo.ClusterName, + TaskName: vmInfo.Name, + } + syncTask(l.svcCtx.DbEngin, noticeInfo) } } - return &resp, nil } -func syncTask(gorm *gorm.DB, taskId int64) { +func syncTask(gorm *gorm.DB, noticeInfo clientCore.NoticeInfo) { var allStatus string - tx := gorm.Raw("SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.status) ,GROUP_CONCAT(DISTINCT a.status) ,GROUP_CONCAT(DISTINCT c.status))as status from task t left join task_hpc h on t.id = h.task_id left join task_cloud c on t.id = c.task_id left join task_ai a on t.id = a.task_id where t.id = ?", taskId).Scan(&allStatus) + tx := gorm.Raw("SELECT CONCAT_WS(',',GROUP_CONCAT(DISTINCT h.status) ,GROUP_CONCAT(DISTINCT a.status) ,GROUP_CONCAT(DISTINCT c.status))as status from task t left join task_hpc h on t.id = h.task_id left join task_cloud c on t.id = c.task_id left join task_ai a on t.id = a.task_id where t.id = ?", noticeInfo.TaskId).Scan(&allStatus) if tx.Error != nil { logx.Error(tx.Error) } - for pcmStatus, hpcStatus := range clientCore.HpcStatusMapping { - for _, status := range hpcStatus { + for pcmStatus, ProviderStatus := range clientCore.StatusMapping { + for _, originalStatus := range ProviderStatus { // if Failed type status appears in subTask then update mainTask to Failed - if pcmStatus == "Failed" && strings.Contains(allStatus, status) { - updateTask(gorm, taskId, constants.Failed) + if pcmStatus == "Failed" && strings.Contains(allStatus, originalStatus) { + updateTask(gorm, noticeInfo.TaskId, constants.Failed) + noticeInfo := clientCore.NoticeInfo{ + AdapterId: noticeInfo.AdapterId, + AdapterName: noticeInfo.AdapterName, + ClusterId: noticeInfo.ClusterId, + ClusterName: noticeInfo.ClusterName, + NoticeType: "failed", + TaskName: noticeInfo.TaskName, + Incident: "任务执行失败，请查看日志！", + CreatedTime: time.Now(), + } + gorm.Table("t_notice").Create(¬iceInfo) return // no Failed type status in subTask,if Saved type status appears in subTask then update mainTask to Saved - } else if pcmStatus == "Saved" { - if strings.Contains(allStatus, status) { - updateTask(gorm, taskId, constants.Saved) + } else if pcmStatus == "Saved" && strings.Contains(allStatus, originalStatus) { + if getTaskStatus(gorm, noticeInfo.TaskId) != "Saved" { + updateTask(gorm, noticeInfo.TaskId, constants.Saved) + noticeInfo := clientCore.NoticeInfo{ + AdapterId: noticeInfo.AdapterId, + AdapterName: noticeInfo.AdapterName, + ClusterId: noticeInfo.ClusterId, + ClusterName: noticeInfo.ClusterName, + NoticeType: "saved", + TaskName: noticeInfo.TaskName, + Incident: "任务已处于队列中！", + CreatedTime: time.Now(), + } + gorm.Table("t_notice").Create(¬iceInfo) + return + } else { return } // no Failed and Saved type status in subTask,if Running type status appears in subTask then update mainTask to Running - } else if pcmStatus == "Running" { - if strings.Contains(allStatus, status) { - updateTask(gorm, taskId, constants.Running) + } else if pcmStatus == "Running" && strings.Contains(allStatus, originalStatus) { + if getTaskStatus(gorm, noticeInfo.TaskId) != "Running" { + updateTask(gorm, noticeInfo.TaskId, constants.Running) + noticeInfo := clientCore.NoticeInfo{ + AdapterId: noticeInfo.AdapterId, + AdapterName: noticeInfo.AdapterName, + ClusterId: noticeInfo.ClusterId, + ClusterName: noticeInfo.ClusterName, + NoticeType: "running", + TaskName: noticeInfo.TaskName, + Incident: "任务状态切换为运行中！", + CreatedTime: time.Now(), + } + gorm.Table("t_notice").Create(¬iceInfo) + return + } else { return } + // at last, mainTask should be succeeded } else { - if strings.Contains(allStatus, status) { - updateTask(gorm, taskId, constants.Succeeded) + if strings.Contains(allStatus, originalStatus) { + updateTask(gorm, noticeInfo.TaskId, constants.Succeeded) + noticeInfo := clientCore.NoticeInfo{ + AdapterId: noticeInfo.AdapterId, + AdapterName: noticeInfo.AdapterName, + ClusterId: noticeInfo.ClusterId, + ClusterName: noticeInfo.ClusterName, + NoticeType: "succeeded", + TaskName: noticeInfo.TaskName, + Incident: "任务执行完成！", + CreatedTime: time.Now(), + } + gorm.Table("t_notice").Create(¬iceInfo) return } } @@ -117,22 +199,10 @@ func updateTask(gorm *gorm.DB, taskId int64, status string) { } gorm.Updates(&task) } - } -func removeRepeatedElement(arr []string) (newArr []string) { - newArr = make([]string, 0) - for i := 0; i < len(arr); i++ { - repeat := false - for j := i + 1; j < len(arr); j++ { - if arr[i] == arr[j] { - repeat = true - break - } - } - if !repeat { - newArr = append(newArr, arr[i]) - } - } - return +func getTaskStatus(gorm *gorm.DB, taskId int64) (status string) { + var task models.Task + gorm.Where("id = ? ", taskId).Find(&task) + return task.Status } diff --git a/api/internal/logic/hpc/commithpctasklogic.go b/api/internal/logic/hpc/commithpctasklogic.go index b763a068..994f3f91 100644 --- a/api/internal/logic/hpc/commithpctasklogic.go +++ b/api/internal/logic/hpc/commithpctasklogic.go @@ -52,6 +52,13 @@ func (l *CommitHpcTaskLogic) CommitHpcTask(req *types.CommitHpcTaskReq) (resp *t var clusterIds []int64 l.svcCtx.DbEngin.Raw("SELECT id FROM `t_cluster` where adapter_id = ? and label = ?", req.AdapterId, req.ClusterType).Scan(&clusterIds) + adapterId, _ := strconv.ParseInt(req.AdapterId, 10, 64) + var adapterName string + l.svcCtx.DbEngin.Raw("SELECT name FROM `t_adapter` where id = ?", req.AdapterId).Scan(&adapterName) + clusterId := clusterIds[rand.Intn(len(clusterIds))] + var clusterName string + l.svcCtx.DbEngin.Raw("SELECT nickname FROM `t_cluster` where id = ?", clusterId).Scan(&clusterName) + env, _ := json.Marshal(req.Environment) if len(clusterIds) == 0 || clusterIds == nil { @@ -62,7 +69,10 @@ func (l *CommitHpcTaskLogic) CommitHpcTask(req *types.CommitHpcTaskReq) (resp *t hpcInfo := models.TaskHpc{ TaskId: taskModel.Id, - ClusterId: clusterIds[rand.Intn(len(clusterIds))], + AdapterId: uint(adapterId), + AdapterName: adapterName, + ClusterId: uint(clusterId), + ClusterName: clusterName, Name: taskModel.Name, Status: "Saved", CmdScript: req.CmdScript, @@ -90,12 +100,11 @@ func (l *CommitHpcTaskLogic) CommitHpcTask(req *types.CommitHpcTaskReq) (resp *t if tx.Error != nil { return nil, tx.Error } - adapterId, _ := strconv.ParseUint(req.AdapterId, 10, 64) - adapterName := "" - tx.Table("t_adapter").Select("name").Where("id=?", adapterId).Find(&adapterName) noticeInfo := clientCore.NoticeInfo{ - AdapterId: int64(adapterId), + AdapterId: adapterId, AdapterName: adapterName, + ClusterId: clusterId, + ClusterName: clusterName, NoticeType: "create", TaskName: req.Name, Incident: "任务创建中", diff --git a/api/internal/scheduler/schedulers/aiScheduler.go b/api/internal/scheduler/schedulers/aiScheduler.go index 76b1a6b9..b61e03b7 100644 --- a/api/internal/scheduler/schedulers/aiScheduler.go +++ b/api/internal/scheduler/schedulers/aiScheduler.go @@ -55,10 +55,10 @@ func NewAiScheduler(ctx context.Context, val string, scheduler *scheduler.Schedu func (as *AiScheduler) GetNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) { ai := models.Ai{ - ParticipantId: participantId, - TaskId: task.TaskId, - Status: "Saved", - YamlString: as.yamlString, + AdapterId: participantId, + TaskId: task.TaskId, + Status: "Saved", + YamlString: as.yamlString, } utils.Convert(task.Metadata, &ai) return ai, nil diff --git a/pkg/models/aimodel_gen.go b/pkg/models/aimodel_gen.go index 160c56fc..7736bc5f 100644 --- a/pkg/models/aimodel_gen.go +++ b/pkg/models/aimodel_gen.go @@ -36,19 +36,22 @@ type ( } Ai struct { - Id int64 `db:"id"` // id - TaskId int64 `db:"task_id"` // 任务id - ParticipantId int64 `db:"participant_id"` // 集群静态信息id - ProjectId string `db:"project_id"` // 项目id - Name string `db:"name"` // 名称 - Status string `db:"status"` // 状态 - StartTime string `db:"start_time"` // 开始时间 - RunningTime int64 `db:"running_time"` // 运行时间 - CreatedBy int64 `db:"created_by"` // 创建人 - CreatedTime sql.NullTime `db:"created_time"` // 创建时间 - UpdatedBy int64 `db:"updated_by"` // 更新人 - UpdatedTime sql.NullTime `db:"updated_time"` // 更新时间 - DeletedFlag int64 `db:"deleted_flag"` // 是否删除（0-否，1-是） + Id int64 `db:"id"` // id + TaskId int64 `db:"task_id"` // 任务id + AdapterId int64 `db:"adapter_id"` // 适配器id + AdapterName string `db:"adapter_name"` //适配器名称 + ClusterId int64 `db:"cluster_id"` //集群id + ClusterName string `db:"cluster_name"` //集群名称 + ProjectId string `db:"project_id"` // 项目id + Name string `db:"name"` // 名称 + Status string `db:"status"` // 状态 + StartTime string `db:"start_time"` // 开始时间 + RunningTime int64 `db:"running_time"` // 运行时间 + CreatedBy int64 `db:"created_by"` // 创建人 + CreatedTime sql.NullTime `db:"created_time"` // 创建时间 + UpdatedBy int64 `db:"updated_by"` // 更新人 + UpdatedTime sql.NullTime `db:"updated_time"` // 更新时间 + DeletedFlag int64 `db:"deleted_flag"` // 是否删除（0-否，1-是） Result string `db:"result"` YamlString string `db:"yaml_string"` JobId string `db:"job_id"` diff --git a/pkg/models/cloud/task_cloud.go b/pkg/models/cloud/task_cloud.go index d60c236e..5cb1e1c9 100644 --- a/pkg/models/cloud/task_cloud.go +++ b/pkg/models/cloud/task_cloud.go @@ -9,6 +9,7 @@ type TaskCloudModel struct { Id uint `json:"id" gorm:"primarykey;not null;comment:id"` TaskId uint `json:"taskId" gorm:"not null;comment:task表id"` AdapterId uint `json:"adapterId" gorm:"not null;comment:适配器id"` + AdapterName string `json:"adapterName" gorm:"not null;comment:适配器名称"` ClusterId uint `json:"clusterId" gorm:"not null;comment:集群id"` ClusterName string `json:"clusterName" gorm:"not null;comment:集群名称"` Kind string `json:"kind" gorm:"comment:种类"` diff --git a/pkg/models/taskhpcmodel_gen.go b/pkg/models/taskhpcmodel_gen.go index 9e2fc1cb..05b41aca 100644 --- a/pkg/models/taskhpcmodel_gen.go +++ b/pkg/models/taskhpcmodel_gen.go @@ -36,12 +36,15 @@ type ( } TaskHpc struct { - Id int64 `db:"id"` // id - TaskId int64 `db:"task_id"` // 任务id - JobId string `db:"job_id"` // 作业id(在第三方系统中的作业id) - ClusterId int64 `db:"cluster_id"` // 执行任务的集群id - Name string `db:"name"` // 名称 - Status string `db:"status"` // 状态 + Id int64 `db:"id"` // id + TaskId int64 `db:"task_id"` // 任务id + JobId string `db:"job_id"` // 作业id(在第三方系统中的作业id) + AdapterId uint `db:"adapter_d"` // 适配器id + AdapterName string `db:"adapter_name"` //适配器名称 + ClusterId uint `db:"cluster_id"` //集群id + ClusterName string `db:"cluster_name"` //集群名称 + Name string `db:"name"` // 名称 + Status string `db:"status"` // 状态 CmdScript string `db:"cmd_script"` StartTime string `db:"start_time"` // 开始时间 RunningTime int64 `db:"running_time"` // 运行时间 diff --git a/pkg/models/taskvmmodel_gen.go b/pkg/models/taskvmmodel_gen.go index 6a89501a..6b147b68 100644 --- a/pkg/models/taskvmmodel_gen.go +++ b/pkg/models/taskvmmodel_gen.go @@ -39,8 +39,10 @@ type ( ParticipantId int64 `db:"participant_id"` // p端id TaskId int64 `db:"task_id"` // 任务id Name string `db:"name"` // 虚拟机名称 - AdapterId int64 `db:"adapter_id"` // 执行任务的适配器id - ClusterId int64 `db:"cluster_id"` // 执行任务的集群id + AdapterId int64 `db:"adapter_id"` // 适配器id + AdapterName string `db:"adapter_name"` //适配器名称 + ClusterId int64 `db:"cluster_id"` //集群id + ClusterName string `db:"cluster_name"` //集群名称 FlavorRef string `db:"flavor_ref"` // 规格索引 ImageRef string `db:"image_ref"` // 镜像索引 Status string `db:"status"` // 状态 From 7c41605443dc6734e1b011a806b7d0230e7aaecd Mon Sep 17 00:00:00 2001 From: Jake <450705171@qq.com> Date: Fri, 10 May 2024 22:16:23 +0800 Subject: [PATCH 20/20] merge conflict Former-commit-id: 6bb70e13e9ec954eb128dd3d8c8201d9cbc54529 --- pkg/models/cloud/task_cloud.go | 1 - 1 file changed, 1 deletion(-) diff --git a/pkg/models/cloud/task_cloud.go b/pkg/models/cloud/task_cloud.go index 784196c0..5cb1e1c9 100644 --- a/pkg/models/cloud/task_cloud.go +++ b/pkg/models/cloud/task_cloud.go @@ -8,7 +8,6 @@ import ( type TaskCloudModel struct { Id uint `json:"id" gorm:"primarykey;not null;comment:id"` TaskId uint `json:"taskId" gorm:"not null;comment:task表id"` - AdapterName string `json:"adapterName" gorm:"not null;comment:适配器名称"` AdapterId uint `json:"adapterId" gorm:"not null;comment:适配器id"` AdapterName string `json:"adapterName" gorm:"not null;comment:适配器名称"` ClusterId uint `json:"clusterId" gorm:"not null;comment:集群id"`