fix: imporve task query speed (#21921)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

pkg/cloudcommon/consts/db.go

@@ -30,6 +30,8 @@ var (
 	taskWorkerCount      int
 	localTaskWorkerCount int
 
+	taskArchiveThresholdHours int
+
 	enableChangeOwnerAutoRename = false
 
 	enableDefaultPolicy = true
@@ -84,10 +86,18 @@ func GetChangeOwnerAutoRename() bool {
 	return enableChangeOwnerAutoRename
 }
 
+func SetTaskArchiveThresholdHours(hours int) {
+	taskArchiveThresholdHours = hours
+}
+
 func TaskWorkerCount() int {
 	return taskWorkerCount
 }
 
 func LocalTaskWorkerCount() int {
 	return localTaskWorkerCount
 }
+
+func TaskArchiveThresholdHours() int {
+	return taskArchiveThresholdHours
+}

pkg/cloudcommon/db/db_dispatcher.go

@@ -583,18 +583,9 @@ func ListItems(manager IModelManager, ctx context.Context, userCred mcclient.Tok
 		limit = exportLimit
 	}
 
-	var (
-		q           *sqlchemy.SQuery
-		useRawQuery bool
-	)
-	{
-		// query senders are responsible for clear up other constraint
-		// like setting "pendinge_delete" to "all"
-		queryDelete, _ := query.GetString("delete")
-		if queryDelete == "all" && policy.PolicyManager.Allow(rbacscope.ScopeSystem, userCred, consts.GetServiceType(), manager.KeywordPlural(), policy.PolicyActionList).Result.IsAllow() {
-			useRawQuery = true
-		}
-	}
+	var q *sqlchemy.SQuery
+
+	useRawQuery := isRawQuery(manager, userCred, query, policy.PolicyActionList)
 
 	queryDict, ok := query.(*jsonutils.JSONDict)
 	if !ok {

pkg/cloudcommon/db/fetch.go

@@ -143,8 +143,22 @@ func FetchByIdOrName(ctx context.Context, manager IModelManager, userCred mcclie
 	}
 }
 
-func fetchItemById(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, idStr string, query jsonutils.JSONObject) (IModel, error) {
-	q := manager.Query()
+func isRawQuery(manager IModelManager, userCred mcclient.TokenCredential, query jsonutils.JSONObject, action string) bool {
+	if query == nil || !query.Contains("delete") {
+		return false
+	}
+	var useRawQuery bool
+	// query senders are responsible for clear up other constraint
+	// like setting "pendinge_delete" to "all"
+	queryDelete, _ := query.GetString("delete")
+	if queryDelete == "all" && policy.PolicyManager.Allow(rbacscope.ScopeSystem, userCred, consts.GetServiceType(), manager.KeywordPlural(), action).Result.IsAllow() {
+		useRawQuery = true
+	}
+	return useRawQuery
+}
+
+func fetchItemById(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, idStr string, query jsonutils.JSONObject, useRawQuery bool) (IModel, error) {
+	q := manager.NewQuery(ctx, userCred, query, useRawQuery)
 	var err error
 	if query != nil && !query.IsZero() {
 		// if isListRbacAllowed(manager, userCred, true) {
@@ -177,8 +191,8 @@ func fetchItemById(manager IModelManager, ctx context.Context, userCred mcclient
 	}
 }
 
-func fetchItemByName(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, idStr string, query jsonutils.JSONObject) (IModel, error) {
-	q := manager.Query()
+func fetchItemByName(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, idStr string, query jsonutils.JSONObject, useRawQuery bool) (IModel, error) {
+	q := manager.NewQuery(ctx, userCred, query, useRawQuery)
 	var err error
 	if query != nil && !query.IsZero() {
 		q, err = listItemQueryFilters(manager, ctx, q, userCred, query, policy.PolicyActionGet, false)
@@ -224,9 +238,10 @@ func fetchItemByName(manager IModelManager, ctx context.Context, userCred mcclie
 }
 
 func fetchItem(manager IModelManager, ctx context.Context, userCred mcclient.TokenCredential, idStr string, query jsonutils.JSONObject) (IModel, error) {
-	item, err := fetchItemById(manager, ctx, userCred, idStr, query)
+	useRawQuery := isRawQuery(manager, userCred, query, policy.PolicyActionGet)
+	item, err := fetchItemById(manager, ctx, userCred, idStr, query, useRawQuery)
 	if err != nil {
-		item, err = fetchItemByName(manager, ctx, userCred, idStr, query)
+		item, err = fetchItemByName(manager, ctx, userCred, idStr, query, useRawQuery)
 	}
 	if err != nil {
 		return nil, err

pkg/cloudcommon/db/taskman/subtasks.go

@@ -71,7 +71,10 @@ func (manager *SSubTaskmanager) GetSubTask(ptaskId string, subtaskId string) *SS
 }
 
 func (manager *SSubTaskmanager) getTotalSubtasksQuery(taskId string, stage string, status string) *sqlchemy.SQuery {
-	q := manager.Query().Equals("task_id", taskId).Equals("stage", stage)
+	q := manager.Query().Equals("task_id", taskId)
+	if len(stage) > 0 {
+		q = q.Equals("stage", stage)
+	}
 	if len(status) > 0 {
 		q = q.Equals("status", status)
 	}

pkg/cloudcommon/db/taskman/taskarchive.go

@@ -0,0 +1,101 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package taskman
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/rbacscope"
+	"yunion.io/x/sqlchemy"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient"
+)
+
+type SArchivedTaskManager struct {
+	db.SLogBaseManager
+	db.SStatusResourceBaseManager
+}
+
+var ArchivedTaskManager *SArchivedTaskManager
+
+func InitArchivedTaskManager() {
+	ArchivedTaskManager = &SArchivedTaskManager{
+		SLogBaseManager: db.NewLogBaseManager(
+			SArchivedTask{},
+			"archived_tasks_tbl",
+			"achivedtask",
+			"achivedtasks",
+			"start_at",
+			consts.OpsLogWithClickhouse,
+		),
+	}
+	ArchivedTaskManager.SetVirtualObject(ArchivedTaskManager)
+}
+
+type SArchivedTask struct {
+	db.SLogBase
+
+	TaskId string `width:"36" charset:"ascii" index:"true" list:"user"`
+
+	STaskBase
+
+	ObjIds     []string `charset:"ascii" list:"user"`
+	ObjNames   []string `charset:"utf8" list:"user"`
+	ProjectIds []string `charset:"ascii" list:"user"`
+	DomainIds  []string `charset:"ascii" list:"user"`
+
+	SubTaskCount   int `json:"sub_task_count" list:"user"`
+	FailSubTaskCnt int `json:"fail_sub_task_cnt" list:"user"`
+	SuccSubTaskCnt int `json:"succ_sub_task_cnt" list:"user"`
+}
+
+func (manager *SArchivedTaskManager) NamespaceScope() rbacscope.TRbacScope {
+	return rbacscope.ScopeProject
+}
+
+func (manager *SArchivedTaskManager) Insert(ctx context.Context, task *STask) error {
+	archivedTask := SArchivedTask{}
+	archivedTask.TaskId = task.Id
+	archivedTask.STaskBase = task.STaskBase
+	archivedTask.ObjIds = TaskObjectManager.GetObjectIds(task)
+	archivedTask.ObjNames = TaskObjectManager.GetObjectNames(task)
+	archivedTask.FailSubTaskCnt, _ = SubTaskManager.GetSubtasksCount(task.Id, "", SUBTASK_FAIL)
+	archivedTask.SuccSubTaskCnt, _ = SubTaskManager.GetSubtasksCount(task.Id, "", SUBTASK_SUCC)
+	archivedTask.SubTaskCount = archivedTask.FailSubTaskCnt + archivedTask.SuccSubTaskCnt
+	archivedTask.ProjectIds = TaskObjectManager.GetProjectIds(task)
+	archivedTask.DomainIds = TaskObjectManager.GetDomainIds(task)
+
+	err := manager.TableSpec().Insert(ctx, &archivedTask)
+	if err != nil {
+		return errors.Wrap(err, "Insert")
+	}
+	return nil
+}
+
+func (task *SArchivedTask) GetOwnerId() mcclient.IIdentityProvider {
+	return nil
+}
+
+func (manager *SArchivedTaskManager) FilterByOwner(ctx context.Context, q *sqlchemy.SQuery, man db.FilterByOwnerProvider, userCred mcclient.TokenCredential, owner mcclient.IIdentityProvider, scope rbacscope.TRbacScope) *sqlchemy.SQuery {
+	return q
+}
+
+func (manager *SArchivedTaskManager) FetchOwnerId(ctx context.Context, data jsonutils.JSONObject) (mcclient.IIdentityProvider, error) {
+	return nil, nil
+}

pkg/cloudcommon/db/taskman/taskbase.go

@@ -0,0 +1,47 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package taskman
+
+import (
+	"time"
+
+	"yunion.io/x/jsonutils"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient"
+)
+
+type STaskBase struct {
+	db.SStatusResourceBase
+
+	// 开始任务时间
+	StartAt time.Time `nullable:"true" list:"user" json:"start_at"`
+	// 完成任务时间
+	EndAt time.Time `nullable:"true" list:"user" json:"end_at"`
+
+	ObjType  string `old_name:"obj_name" json:"obj_type" width:"128" charset:"utf8" nullable:"true" list:"user"`
+	Object   string `json:"object" width:"128" charset:"utf8" nullable:"true" list:"user"`  //  Column(VARCHAR(128, charset='utf8'), nullable=False)
+	ObjId    string `width:"128" charset:"ascii" nullable:"false" list:"user" index:"true"` // Column(VARCHAR(ID_LENGTH, charset='ascii'), nullable=False)
+	TaskName string `width:"64" charset:"ascii" nullable:"false" list:"user" index:"true"`  // Column(VARCHAR(64, charset='ascii'), nullable=False)
+
+	UserCred mcclient.TokenCredential `width:"1024" charset:"utf8" nullable:"false" get:"user"` // Column(VARCHAR(1024, charset='ascii'), nullable=False)
+	// OwnerCred string `width:"512" charset:"ascii" nullable:"true"` // Column(VARCHAR(512, charset='ascii'), nullable=True)
+	Params *jsonutils.JSONDict `charset:"utf8" length:"medium" nullable:"false" get:"user"` // Column(MEDIUMTEXT(charset='ascii'), nullable=False)
+
+	Stage string `width:"64" charset:"ascii" nullable:"false" default:"on_init" list:"user" index:"true"` // Column(VARCHAR(64, charset='ascii'), nullable=False, default='on_init')
+
+	// 父任务Id
+	ParentTaskId string `width:"36" charset:"ascii" list:"user" index:"true" json:"parent_task_id"`
+}

pkg/cloudcommon/db/taskman/taskobjs.go

@@ -55,18 +55,18 @@ type STaskObject struct {
 	db.SProjectizedResourceBase
 
 	TaskId string `width:"36" charset:"ascii" nullable:"false" primary:"true" index:"true"` // Column(VARCHAR(36, charset='ascii'), nullable=False, primary_key=True, index=True)
-	ObjId  string `width:"36" charset:"ascii" nullable:"false" primary:"true"`              // Column(VARCHAR(36, charset='ascii'), nullable=False, primary_key=True)
+	ObjId  string `width:"128" charset:"ascii" nullable:"false" primary:"true"`             // Column(VARCHAR(36, charset='ascii'), nullable=False, primary_key=True)
 	Object string `json:"object" width:"128" charset:"utf8" nullable:"false" list:"user"`
 }
 
-func (manager *STaskObjectManager) GetObjectIds(task *STask) []string {
+func (manager *STaskObjectManager) getValues(task *STask, field string) []string {
 	ret := make([]string, 0)
 	taskobjs := manager.Query().SubQuery()
-	q := taskobjs.Query(taskobjs.Field("obj_id")).Equals("task_id", task.Id)
+	q := taskobjs.Query(taskobjs.Field(field)).Equals("task_id", task.Id).Distinct()
 	rows, err := q.Rows()
 	if err != nil {
 		if err != sql.ErrNoRows {
-			log.Errorf("TaskObjectManager GetObjectIds fail %s", err)
+			log.Errorf("TaskObjectManager getValues fail %s", err)
 		}
 		return nil
 	}
@@ -75,36 +75,28 @@ func (manager *STaskObjectManager) GetObjectIds(task *STask) []string {
 		var objId string
 		err = rows.Scan(&objId)
 		if err != nil {
-			log.Errorf("TaskObjectManager GetObjects fetch row fail %s", err)
+			log.Errorf("TaskObjectManager getValues fetch row fail %s", err)
 			return nil
 		}
 		ret = append(ret, objId)
 	}
 	return ret
 }
 
+func (manager *STaskObjectManager) GetObjectIds(task *STask) []string {
+	return manager.getValues(task, "obj_id")
+}
+
 func (manager *STaskObjectManager) GetObjectNames(task *STask) []string {
-	ret := make([]string, 0)
-	taskobjs := manager.Query().SubQuery()
-	q := taskobjs.Query(taskobjs.Field("object")).Equals("task_id", task.Id)
-	rows, err := q.Rows()
-	if err != nil {
-		if err != sql.ErrNoRows {
-			log.Errorf("TaskObjectManager GetObjectIds fail %s", err)
-		}
-		return nil
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var objId string
-		err = rows.Scan(&objId)
-		if err != nil {
-			log.Errorf("TaskObjectManager GetObjects fetch row fail %s", err)
-			return nil
-		}
-		ret = append(ret, objId)
-	}
-	return ret
+	return manager.getValues(task, "object")
+}
+
+func (manager *STaskObjectManager) GetProjectIds(task *STask) []string {
+	return manager.getValues(task, "tenant_id")
+}
+
+func (manager *STaskObjectManager) GetDomainIds(task *STask) []string {
+	return manager.getValues(task, "domain_id")
 }
 
 func (manager *STaskObjectManager) FetchOwnerId(ctx context.Context, data jsonutils.JSONObject) (mcclient.IIdentityProvider, error) {
@@ -148,3 +140,22 @@ func (manager *STaskObjectManager) ResourceScope() rbacscope.TRbacScope {
 func (taskObj *STaskObject) GetOwnerId() mcclient.IIdentityProvider {
 	return taskObj.SProjectizedResourceBase.GetOwnerId()
 }
+
+func (manager *STaskObjectManager) insertObject(ctx context.Context, taskId string, obj db.IStandaloneModel) (*STaskObject, error) {
+	to := STaskObject{
+		TaskId: taskId,
+		ObjId:  obj.GetId(),
+		Object: obj.GetName(),
+	}
+	ownerId := obj.GetOwnerId()
+	if ownerId != nil {
+		to.DomainId = ownerId.GetProjectDomainId()
+		to.ProjectId = ownerId.GetProjectId()
+	}
+	to.SetModelManager(TaskObjectManager, &to)
+	err := TaskObjectManager.TableSpec().Insert(ctx, &to)
+	if err != nil {
+		return nil, errors.Wrap(err, "insert task object")
+	}
+	return &to, nil
+}