pcm-coordinator/rpc/internal/cron/cronlogic.go

/*

 Copyright (c) [2023] [pcm]
 [pcm-coordinator] is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
 THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 EITHER EXPaRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.

*/

package cron

import (
	"github.com/zeromicro/go-zero/core/logx"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/constants"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/models"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/rpc/internal/svc"
	"gorm.io/gorm"
	"strings"
)

func InitCron(svc *svc.ServiceContext) {
	svc.Cron.Start()
	svc.Cron.AddFunc("*/5  *  *  *  *  ?", func() {

		var tasks []models.Task
		svc.DbEngin.Where("status not in  ?", []string{constants.Deleted, constants.Succeeded, constants.Completed, constants.Failed}).Find(&tasks)
		for _, task := range tasks {
			var allStatus string
			tx := svc.DbEngin.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 cloud c  on t.id  = c.task_id left join ai a on t.id  = a.task_id where t.id  = ?", task.Id).Scan(&allStatus)
			if tx.Error != nil {
				logx.Error(tx.Error)
			}
			// 子状态统一则修改主任务状态
			statusArray := strings.Split(allStatus, ",")
			if len(removeRepeatedElement(statusArray)) == 1 {
				updateTask(svc.DbEngin, &task, statusArray[0])
				continue
			}
			// 子任务包含失败状态 主任务则失败
			if strings.Contains(allStatus, constants.Failed) {
				updateTask(svc.DbEngin, &task, constants.Failed)
				continue
			}
			if strings.Contains(allStatus, constants.Running) {
				updateTask(svc.DbEngin, &task, constants.Running)
			}
		}

	})
}

func updateTask(dbEngin *gorm.DB, task *models.Task, status string) {
	if task.Status != status {
		task.Status = status
		dbEngin.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
}