ob/agent/executor/task/sub_task.go - master - zhendosina/obshell

ob/agent/executor/task/sub_task.go
184 строки5 KB
Junkrat77
PullRequest: 614 feat: support seekdb
31 окт 2025, 11:06
31 окт 2025, 11:0655d9f3e
/*
 * Copyright (c) 2024 OceanBase.
 *
 * 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 task

import (
	"github.com/gin-gonic/gin"

	"github.com/oceanbase/obshell/ob/agent/api/common"
	"github.com/oceanbase/obshell/ob/agent/engine/task"
	"github.com/oceanbase/obshell/ob/agent/errors"
	"github.com/oceanbase/obshell/ob/agent/meta"
	taskservice "github.com/oceanbase/obshell/ob/agent/service/task"
)

// get sub_task detail by id
//
//	@ID				getSubTaskDetail
//	@Summary		get sub_task detail by sub_task_id
//	@Description	get sub_task detail by sub_task_id
//	@Tags			task
//	@Accept			application/json
//	@Produce		application/json
//	@Param			X-OCS-Header	header	string					true	"Authorization"
//	@Param			id				path	string					true	"id"
//	@Param			showDetails		query	param.TaskQueryParams	true	"show details"
//	@Success		200				object	http.OcsAgentResponse{data=task.TaskDetailDTO}
//	@Failure		400				object	http.OcsAgentResponse
//	@Failure		404				object	http.OcsAgentResponse
//	@Failure		500				object	http.OcsAgentResponse
//	@Router			/api/v1/task/sub_task/{id} [get]
func GetSubTaskDetail(c *gin.Context) {
	var taskDTOParam task.TaskDetailDTO
	var taskDetailDTO *task.TaskDetailDTO
	var service taskservice.TaskServiceInterface

	if err := c.BindUri(&taskDTOParam); err != nil {
		common.SendResponse(c, nil, err)
		return
	}

	taskID, agent, err := task.ConvertGenericID(taskDTOParam.GenericID)
	if err != nil {
		common.SendResponse(c, nil, err)
		return
	}

	if agent != nil && !meta.OCS_AGENT.Equal(agent) {
		if task.IsObproxyTask(taskDTOParam.GenericID) {
			common.SendResponse(c, nil, errors.Occur(errors.ErrTaskNotFoundWithReason, "obproxy task not found"))
		}
		if meta.OCS_AGENT.IsFollowerAgent() {
			// forward request to master
			master := agentService.GetMasterAgentInfo()
			if master == nil {
				common.SendResponse(c, nil, errors.Occur(errors.ErrAgentNoMaster))
				return
			}
			common.ForwardRequest(c, master, nil)
		} else {
			common.ForwardRequest(c, agent, nil)
		}
		return
	}

	if agent == nil {
		service = clusterTaskService
	} else {
		service = localTaskService
	}

	subTask, err := service.GetSubTaskByTaskID(taskID)
	if err != nil {
		common.SendResponse(c, nil, errors.WrapRetain(errors.ErrTaskNotFound, err))
		return
	}

	dagType := ""
	if subTask.IsLocalTask() {
		dag, err := service.GetDagBySubTaskId(subTask.GetID())
		if err != nil {
			common.SendResponse(c, nil, errors.WrapRetain(errors.ErrTaskNotFound, err))
			return
		}
		dagType = dag.GetDagType()
		if task.ConvertToGenericID(dag, dag.GetDagType())[0] != taskDTOParam.GenericID[0] {
			common.SendResponse(c, nil, errors.Occur(errors.ErrTaskNotFoundWithReason, "sub task type not match"))
			return
		}

	}

	taskDetailDTO, err = getSubTaskDetail(service, subTask, dagType)
	if err != nil {
		common.SendResponse(c, nil, err)
		return
	}
	taskDetailDTO.SetVisible(true)
	common.SendResponse(c, taskDetailDTO, nil)
}

func getSubTaskDetail(service taskservice.TaskServiceInterface, subTask task.ExecutableTask, dagType string) (taskDetailDTO *task.TaskDetailDTO, err error) {
	taskDetailDTO = task.NewTaskDetailDTO(subTask, dagType)
	if subTask.IsRunning() || subTask.IsFinished() {
		taskDetailDTO.TaskLogs, err = service.GetSubTaskLogsByTaskID(subTask.GetID())
	}
	return
}

// get full sub_task logs by id
//
//	@ID				getFullSubTaskLogs
//	@Summary		get full sub_task logs by sub_task_id
//	@Description	get full sub_task logs by sub_task_id
//	@Tags			task
//	@Accept			application/json
//	@Produce		application/json
//	@Param			X-OCS-Header	header	string	true	"Authorization"
//	@Param			id				path	string	true	"id"
//	@Success		200				object	http.OcsAgentResponse{data=task.TaskDetailDTO}
//	@Failure		400				object	http.OcsAgentResponse
//	@Failure		404				object	http.OcsAgentResponse
//	@Failure		500				object	http.OcsAgentResponse
//	@Router			/api/v1/task/sub_task/{id}/logs [get]
func GetFullSubTaskLogs(c *gin.Context) {
	var taskDTOParam task.TaskDetailDTO
	var service taskservice.TaskServiceInterface

	if err := c.BindUri(&taskDTOParam); err != nil {
		common.SendResponse(c, nil, err)
		return
	}

	taskID, agent, err := task.ConvertGenericID(taskDTOParam.GenericID)
	if err != nil {
		common.SendResponse(c, nil, err)
		return
	}

	if agent != nil && !meta.OCS_AGENT.Equal(agent) {
		if task.IsObproxyTask(taskDTOParam.GenericID) {
			common.SendResponse(c, nil, errors.Occur(errors.ErrTaskNotFoundWithReason, "obproxy task not found"))
		}
		if meta.OCS_AGENT.IsFollowerAgent() {
			// forward request to master
			master := agentService.GetMasterAgentInfo()
			if master == nil {
				common.SendResponse(c, nil, errors.Occur(errors.ErrAgentNoMaster))
				return
			}
			common.ForwardRequest(c, master, nil)
		} else {
			common.ForwardRequest(c, agent, nil)
		}
		return
	}

	if agent == nil {
		service = clusterTaskService
	} else {
		service = localTaskService
	}

	subTaskLogs, err := service.GetFullSubTaskLogsByTaskID(taskID)
	if err != nil {
		common.SendResponse(c, nil, err)
		return
	}

	common.SendResponse(c, subTaskLogs, nil)
}