floraachy
/
console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add api to restart all failed sub tasks

This commit is contained in:
liugq 2023-12-19 17:42:24 +08:00
parent 0a6d0e5696
commit f053282754
2 changed files with 61 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
plugin/task_manager/api.go
View File

@ -22,6 +22,7 @@ func InitAPI() {
api.HandleAPIMethod(api.GET, "/migration/data/:task_id/info/:index", handler.RequirePermission(handler.getDataMigrationTaskOfIndex, enum.PermissionMigrationTaskRead))
api.HandleAPIMethod(api.GET, "/migration/data/:task_id/logging/:index", handler.RequirePermission(handler.searchIndexLevelTaskLogging, enum.PermissionMigrationTaskRead))
api.HandleAPIMethod(api.GET, "/migration/data/_search_values", handler.RequirePermission(handler.searchTaskFieldValues("cluster_migration"), enum.PermissionMigrationTaskRead))
api.HandleAPIMethod(api.POST, "/migration/data/partition/_restart", handler.RequirePermission(handler.restartAllFailedPartitions, enum.PermissionMigrationTaskWrite))
api.HandleAPIMethod(api.GET, "/comparison/data/_search", handler.RequirePermission(handler.searchTask("cluster_comparison"), enum.PermissionComparisonTaskRead))
api.HandleAPIMethod(api.POST, "/comparison/data", handler.RequirePermission(handler.createDataComparisonTask, enum.PermissionComparisonTaskWrite))

60
plugin/task_manager/migration_api.go
View File

@ -443,3 +443,63 @@ func (h *APIHandler) getMigrationMajorTaskInfo(id string) (taskStats migration_m
}
return taskStats, indexState, nil
}
func (h *APIHandler) restartAllFailedPartitions(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
mustQ := []util.MapStr{
{
"term": util.MapStr{
"metadata.type": util.MapStr{
"value": "cluster_migration",
},
},
},
{
"term": util.MapStr{
"status": util.MapStr{
"value": "error",
},
},
},
}
queryDSL := util.MapStr{
"query": util.MapStr{
"bool": util.MapStr{
"must": mustQ,
},
},
"script": util.MapStr{
"source": fmt.Sprintf("ctx._source['status'] = '%s'", task.StatusRunning),
},
}
body := util.MustToJSONBytes(queryDSL)
err := orm.UpdateBy(&task.Task{}, body)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
//update status of sub task
mustQ[0] = util.MapStr{
"term": util.MapStr{
"metadata.type": util.MapStr{
"value": "index_migration",
},
},
}
queryDSL["script"] = util.MapStr{
"source": fmt.Sprintf("ctx._source['status'] = '%s'", task.StatusReady),
}
body = util.MustToJSONBytes(queryDSL)
err = orm.UpdateBy(&task.Task{}, body)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteAckOKJSON(w)
}