Merge pull request 'updated deployinstance logics' (#274) from tzwang/pcm-coordinator:master into master

Former-commit-id: b858434acdf715fec01e7ef73f53f5bacbd0296b

internal/logic/inference/deployinstancelistlogic.go

@@ -38,16 +38,18 @@ func (l *DeployInstanceListLogic) DeployInstanceList(req *types.DeployInstanceLi
 		return nil, tx.Error
 	}
 
+	go updater.UpdateDeployInstanceStatusBatch(l.svcCtx, list)
+
 	ins := list[0]
 	for i := range list {
-		last, _ := time.Parse(time.RFC3339, ins.UpdateTime)
+		uTime, _ := time.Parse(time.RFC3339, ins.UpdateTime)
 		latest, _ := time.Parse(time.RFC3339, list[i].UpdateTime)
-		if latest.After(last) {
+		if latest.After(uTime) {
 			ins = list[i]
 		}
 	}
 
-	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins)
+	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins, true)
 
 	//count total
 	var total int64

internal/logic/inference/startdeployinstancelistlogic.go

@@ -40,7 +40,7 @@ func (l *StartDeployInstanceListLogic) StartDeployInstanceList(req *types.StartD
 		return nil, errors.New("start instance failed")
 	}
 
-	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins)
+	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins, true)
 
 	return resp, nil
 }

internal/logic/inference/stopdeployinstancelogic.go

@@ -40,7 +40,7 @@ func (l *StopDeployInstanceLogic) StopDeployInstance(req *types.StopDeployInstan
 		return nil, errors.New("stop instance failed")
 	}
 
-	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins)
+	go updater.UpdateDeployInstanceStatus(l.svcCtx, ins, true)
 
 	return resp, nil
 }

internal/scheduler/database/aiStorage.go

@@ -441,7 +441,7 @@ func (s *AiStorage) GetInferDeployInstanceTotalNum() (int32, error) {
 
 func (s *AiStorage) GetInferDeployInstanceRunningNum() (int32, error) {
 	var total int32
-	tx := s.DbEngin.Raw("select count(*) from ai_infer_deploy_instance where `status` = 'running'").Scan(&total)
+	tx := s.DbEngin.Raw("select count(*) from ai_infer_deploy_instance where `status` = 'Running'").Scan(&total)
 	if tx.Error != nil {
 		logx.Errorf(tx.Error.Error())
 		return 0, tx.Error

internal/scheduler/service/updater/deployInstance.go

@@ -9,7 +9,25 @@ import (
 	"strconv"
 )
 
-func UpdateDeployInstanceStatus(svc *svc.ServiceContext, instance *models.AiInferDeployInstance) {
+func UpdateDeployInstanceStatusBatch(svc *svc.ServiceContext, insList []*models.AiInferDeployInstance) {
+	list := make([]*models.AiInferDeployInstance, len(insList))
+	copy(list, insList)
+	for i := len(list) - 1; i >= 0; i-- {
+		if list[i].Status == constants.Running || list[i].Status == constants.Stopped {
+			list = append(list[:i], list[i+1:]...)
+		}
+	}
+
+	if len(list) == 0 {
+		return
+	}
+
+	for _, instance := range list {
+		go UpdateDeployInstanceStatus(svc, instance, false)
+	}
+}
+
+func UpdateDeployInstanceStatus(svc *svc.ServiceContext, instance *models.AiInferDeployInstance, updatetime bool) {
 	amap, found := svc.Scheduler.AiService.InferenceAdapterMap[strconv.FormatInt(instance.AdapterId, 10)]
 	if !found {
 		return
@@ -44,8 +62,11 @@ func UpdateDeployInstanceStatus(svc *svc.ServiceContext, instance *models.AiInfe
 			instance.Status = ins.Status
 		}
 	}
-	err = svc.Scheduler.AiStorages.UpdateInferDeployInstance(instance)
-	if err != nil {
-		return
+
+	if updatetime {
+		err = svc.Scheduler.AiStorages.UpdateInferDeployInstance(instance)
+		if err != nil {
+			return
+		}
 	}
 }