myxce/src/frontend/apiserver/controllers/hosts.go
Rongsong Shen 728496bea0
fix(apiserver): populate hosts-list slot usage counts
xcecli hosts list showed NUM_RUN/NUM_SSUSP/NUM_USUSP as always 0 --
the fields existed on HostInfo and were displayed, but nothing ever
assigned them. processHostData now derives per-host slot usage from
the live job store (exec_hosts + job_status), counting occupied slots
so the numbers are comparable to MAX_SLOT.

Partial fix for gitea#46 / bug-1743: the separate "dispatch stops at
8/16" question needs the reporter's live limits.yaml/pend-reason to
confirm a usergroup/queue slot limit, not addressed here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
2026-09-01 06:37:37 +00:00

657 lines
20 KiB
Go

package controllers
import (
"apiserver/backend"
"apiserver/common"
"apiserver/models"
"apiserver/services"
"apiserver/store"
"apiserver/utils"
logger "apiserver/utils"
"encoding/json"
"errors"
"reflect"
"strings"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
)
type HostHandler struct {
}
// GetHosts godoc
// @Summary Get host information
// @Description Get host information
// @Tags hosts
// @Accept json
// @Produce json
// @Success 200
// @Failure 400 {object} models.Error
// @Router /api/v1/hosts [get]
func (h *HostHandler) GetHostsInfo(ctx *gin.Context) {
logger.Debug("Enter GetAllHosts() function from ", zap.String("client_ip", ctx.ClientIP()))
var fields []string
filters := make([]utils.FilterCondition, 0)
sortOrders := make([]utils.SortOrder, 0)
// ---- check parameters ----
for param := range ctx.Request.URL.Query() {
if !utils.ContainsField(utils.SupportedQueryParams, param) {
err := errors.New("invalid host info query parameters of : " + param)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
// Get query parameters
filterStr := ctx.Query("filter")
fieldsStr := ctx.Query("fields")
sortStr := ctx.Query("sort") // sort=host_name:asc,submit_time:desc
pagetokenStr := ctx.Query("page_token") // page_token=Aafjakjf0023A
limitStr := ctx.Query("limit") // eg: limit=100
logger.Debug("Host query values are:",
zap.String("fieldsStr", fieldsStr),
zap.String("filterStr", filterStr),
zap.String("sortStr", sortStr),
zap.String("limitStr", limitStr),
zap.String("pagetokenStr", pagetokenStr))
// ---- check limit value ----
limit, err := utils.ParseLimit(limitStr)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, err.Error(), nil)
return
}
// ---- check filter ----
if filterStr != "" {
filters = utils.ParseFilters(filterStr)
if filters == nil {
err := errors.New("filter " + filterStr + " has syntax error")
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
//filtercds := make([]utils.Condition, len(filters))
for _, filter := range filters {
// check the filter is supported.
if !utils.ContainsField(models.HOST_SUPPORT_FILTERS, filter.Field) {
err := errors.New("unsupported filter: " + filter.Field)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
var hosts models.HostInfo
filtercds, err := utils.ConvertToNewFilter(filters, &hosts)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
// ---- check fields ----
// get fields and do check, if not set, use default display columns
if fieldsStr != "" {
fields = utils.ParseFields(fieldsStr)
found := false
// verify the fields is in supported list
supportedFields := utils.GetSupportedFields(models.HostInfo{})
logger.Debug("host supportedFields: %v", zap.Any("supportedFields", supportedFields))
for _, field := range fields {
if !utils.ContainsField(supportedFields, field) {
err := errors.New("unsupported query field: " + field)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
// if user not specify host-name in fields, add host_name as default field.
if field == "host_name" {
found = true
}
}
if !found {
logger.Debug("user doesn't specify hostname in query fields, add it.")
fields = append(fields, "host_name")
}
} else {
fields = models.HOST_DEFAULT_COLUMN
}
// ---- check sort ----
// handle sort, when no sortStr specified, need add the primary key of host_info
// table into sortStr as default value
if sortStr == "" {
logger.Debug("sort is not specified, set sort=host_name:asc as default")
sortStr = "host_name:asc"
}
// parse sortStr
sorts, err := utils.ParseSortParam(sortStr, models.HOST_SUPPORT_SORT)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
// ---- check paging ---
if pagetokenStr != "" {
pagetokenStr, err = utils.ParsePageToken(pagetokenStr)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
//the primary key of host_info table host_id should always be added into sortStr if user doesn't specify.
found := false
for _, sortent := range sorts {
if sortent.Column == "host_name" {
found = true
break
}
}
// user not specify the host_name, only append host_name at the end.
if !found {
sorts = append(sorts, utils.SortOrder{
Column: "host_name",
Ascending: true,
})
}
//check all sort column is in fields
err = utils.CheckAllMapKeyInList(sorts, fields)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
sortOrders = sorts
//fmt.Println("filtercds:", filtercds, "sortOrders:", sortOrders, "limit:", limit, "pagetokenStr:", pagetokenStr)
queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, limit, pagetokenStr)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_HOSTINFO_MSG+": "+err.Error(), nil)
return
}
hqueryReply, pagetoken, err := services.QueryData(services.TableHostInfo, queryOpts, fields)
if err != nil {
logger.Error("Failed to query host info.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_HOSTINFO_MSG+": "+err.Error(), nil)
return
}
logger.Debug("the host info query reply:", zap.Any("hqueryReply", hqueryReply), zap.String("pagetoken", pagetoken))
//convert to readable string.
queryResult := processHostData(hqueryReply, pagetoken)
ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult)
}
func selectFields(host interface{}, fields []string) map[string]interface{} {
result := make(map[string]interface{})
v := reflect.ValueOf(host)
t := v.Type()
// Create a map of lowercase field names to actual field names
fieldMap := make(map[string]string)
// Handle embedded BasicHost fields
if t.Kind() == reflect.Struct {
for i := 0; i < t.NumField(); i++ {
field := t.Field(i)
// If it's an embedded struct (BasicHost)
if field.Anonymous {
embedValue := v.Field(i)
embedType := embedValue.Type()
for j := 0; j < embedType.NumField(); j++ {
embedField := embedType.Field(j)
fieldMap[strings.ToLower(embedField.Name)] = embedField.Name
}
}
// Add the field itself
fieldMap[strings.ToLower(field.Name)] = field.Name
}
}
for _, field := range fields {
lowerField := strings.ToLower(field)
if actualField, ok := fieldMap[lowerField]; ok {
var value interface{}
var jsonTag string
// If not found in BasicHost, try the main struct
if v.FieldByName(actualField).IsValid() {
value = v.FieldByName(actualField).Interface()
f, _ := t.FieldByName(actualField)
jsonTag = f.Tag.Get("json")
}
if value != nil {
result[jsonTag] = value
}
}
}
return result
}
func (h HostHandler) GetHostByHostname(ctx *gin.Context) {
logger.Debug("Enter GetHostByHostname() function from ", zap.String("client_ip", ctx.ClientIP()))
// ---- check parameters ----
for param := range ctx.Request.URL.Query() {
if !utils.ContainsField(utils.SupportedQueryParams, param) {
err := errors.New("invalid specific host info query parameters of : " + param)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
hostNamesStr := ctx.Param("hostname")
fieldsStr := ctx.Query("fields")
logger.Debug("origin query string hostname, fieldsStr values are:",
zap.String("hostname", hostNamesStr),
zap.String("fields", fieldsStr))
//add host_name list as filter.
hostNames := strings.Split(hostNamesStr, ",")
mergeHqueryReply := make([]utils.OrderMap, 0)
var page_token = ""
for _, hostname := range hostNames {
hostname = strings.TrimSpace(hostname)
filters := make([]utils.FilterCondition, 0)
sortOrders := make([]utils.SortOrder, 0)
filters = append(filters, utils.FilterCondition{
Field: "host_name",
Operator: "eq",
Value: hostname,
})
//filtercds := make([]utils.Condition, len(filters))
for _, filter := range filters {
// check the filter is supported.
if !utils.ContainsField(models.HOST_SUPPORT_FILTERS, filter.Field) {
err := errors.New("unsupported filter: " + filter.Field)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
}
var host models.HostInfo
var err error
filtercds, err := utils.ConvertToNewFilter(filters, &host)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
var fields []string
if fieldsStr != "" {
fields = utils.ParseFields(fieldsStr)
found := false
// verify the fields is in supported list
supportedFields := utils.GetSupportedFields(models.HostInfo{})
logger.Debug("host supportedFields: %v", zap.Any("supportedFields", supportedFields))
for _, field := range fields {
if !utils.ContainsField(supportedFields, field) {
err := errors.New("unsupported query field: " + field)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
// if user not specify host_name in fields, add host_name as default field.
if field == "host_name" {
found = true
}
}
if !found {
logger.Debug("user doesn't specify host_name in query fields, add it.")
fields = append(fields, "host_name")
}
} else { // default display columns
fields = models.HOST_DEFAULT_COLUMN
}
// add a default order: sort=host_name:asc
sortOrders = append(sortOrders, utils.SortOrder{
Column: "host_name",
Ascending: true,
})
queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, utils.PER_PAGE_LIMIT, "")
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
return
}
hqueryReply, page_token, err := services.QueryData(services.TableHostInfo, queryOpts, fields)
if err != nil {
logger.Error("Failed to query the specific host info.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_HOSTINFO_MSG+": "+err.Error(), nil)
return
}
logger.Debug("specific host query reply:", zap.Any("hqueryReply", hqueryReply), zap.String("pagetoken", page_token))
mergeHqueryReply = append(mergeHqueryReply, hqueryReply...)
} // end for hostname
logger.Debug("the specific host query reply:", zap.Any("hqueryReply", mergeHqueryReply))
//convert host_status to readable string.
queryResult := processHostData(mergeHqueryReply, page_token)
ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult)
}
func preCheckHostRequest(req *models.ResReqSelectRequest) error {
if strings.TrimSpace(req.Select) != "" {
if len(req.Select) > 4094 {
return errors.New("Select string exceeds the limit of 4094 characters")
}
} else {
return errors.New("Select string cannot be empty")
}
return nil
}
func (h *HostHandler) FilterHostByResReqSelect(ctx *gin.Context) {
logger.Debug("Enter FilterHostByResReqSelect() function")
// Resource requirement select filtering is not yet implemented without backend.
queryResult := processHostData([]utils.OrderMap{}, "")
ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult)
}
// ControlHost
func (h *HostHandler) ControlHost(ctx *gin.Context) {
logger.Debug("Enter ControlHost() function")
var req models.HostControlRequest
// use strict json format check
if err := utils.BindJSONStrict(ctx, &req); err != nil {
logger.Error("Invalid request parameters", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
allFlag, err := preCheckHostControlRequest(&req)
if err != nil {
logger.Error("Failed to send host control request", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
baseAction := req.ControlAction
logger.Debug("ControlHost baseAction is ", zap.Any("baseAction", baseAction))
// get the request submit user
token := ctx.GetHeader("Authorization")
_, err = GetUsernameFromToken(token)
if err != nil {
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
hostActionMap := map[string]int{
"open": models.HOST_OPEN,
"close": models.HOST_CLOSE,
}
actionType, exists := hostActionMap[baseAction]
if !exists {
logger.Error("Unsupported action type", zap.String("baseAction", baseAction))
ReturnError(ctx, common.INVALID_PARAMS_CODE, "Unsupported action type: "+baseAction, nil)
return
}
successHosts := []map[string]interface{}{}
failedHosts := []map[string]interface{}{}
validatedSuccessHosts, validatedFailedHosts := validateHost(ctx, actionType, req.Hostname)
if ctx.IsAborted() {
ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil)
return
}
if allFlag {
hostnames := store.GetAllHostNames()
validatedSuccessHosts, validatedFailedHosts = validateHost(ctx, actionType, hostnames)
}
for _, hostname := range validatedSuccessHosts {
successHosts = append(successHosts, map[string]interface{}{
"host_name": hostname,
"msg": "host is being controlled",
})
}
for hostname, reason := range validatedFailedHosts {
failedHosts = append(failedHosts, map[string]interface{}{
"host_name": hostname,
"msg": reason,
})
}
for _, host := range successHosts {
hostname := host["host_name"].(string)
err = backend.ControlHost(hostname, actionType)
if err != nil {
logger.Error("Failed to send host control", zap.Error(err))
failedHosts = append(failedHosts, map[string]interface{}{
"host_name": hostname,
"msg": "failed to control host: " + err.Error(),
})
}
}
response, successMsg := BuildBatchOperationResponse(baseAction, "Host", successHosts, failedHosts)
overallStatus := response["status"].(string)
logger.Info("Finished processing host control request",
zap.String("action", baseAction),
zap.String("overallStatus", overallStatus),
zap.Int("succeededCount", len(successHosts)),
zap.Int("failedCount", len(failedHosts)),
zap.String("client_ip", ctx.ClientIP()))
ReturnSuccess(ctx, common.ACCEPTED, successMsg, response)
return
}
// validateHost validates the hostname in the request
func validateHost(ctx *gin.Context, control int, hostnames []string) (successHosts []string, failedHosts map[string]string) {
logger.Debug("Validating hostnames: %v", zap.Any("hostnames", hostnames))
// get logon user
token := ctx.GetHeader("Authorization")
logonUser, err := GetUsernameFromToken(token)
if err != nil {
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
successHosts = []string{}
failedHosts = make(map[string]string)
if len(hostnames) == 0 {
logger.Error("Hostnames cannot be empty")
return
}
for _, hostname := range hostnames {
if hostname == "" {
logger.Error("Hostname cannot be empty")
continue
}
hostPermission, err := GetRequireResourcePermission(logonUser, "host:control")
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil)
return
}
if hostPermission.DomainType == models.DomainTypeHostGroup {
// TODO: check hostgroup values
}
if store.Hosts.Has(hostname) {
logger.Debug("Found hostname in store", zap.String("hostname", hostname))
successHosts = append(successHosts, hostname)
/* no need to check host_status
if rawStatus, exists := hostInfo[0].Data["host_status"]; exists {
if hostStatus, ok := rawStatus.(int64); ok {
db_host_status := int(hostStatus)
if isHostControlActionAllowed(control, db_host_status) {
successHosts = append(successHosts, hostname)
} else {
logger.Error("action not allowed for host", zap.String("hostname", hostname))
failedHosts[hostname] = "action not allowed for current host_status"
}
} else {
logger.Error("host_status is not an int64", zap.Any("host_status", rawStatus))
failedHosts[hostname] = "invalid host_status type for host"
}
} else {
logger.Error("host_status not found in hostInfo", zap.String("hostname", hostname))
failedHosts[hostname] = "host_status not found for host"
} */
} else {
logger.Error("invalid hostname", zap.String("hostname", hostname))
failedHosts[hostname] = common.ERROR_HOST_NOT_FOUND_MSG
}
}
return
}
func isHostControlActionAllowed(actionType int, status int) bool {
isClosed := (status&models.HOSTSTAT["CLOSED"]) != 0 ||
(status&models.HOSTSTAT["CLOSED_BY_ADMIN"]) != 0
switch actionType {
case 1: // open
return isClosed
case 2: // close
return !isClosed
default:
return false
}
}
func preCheckHostControlRequest(req *models.HostControlRequest) (bool, error) {
logger.Debug("Enter preCheckHostControlRequest")
if err := utils.ValidateControlAction(req.ControlAction, models.SupportedHostActions); err != nil {
return false, err
}
req.ControlAction = strings.ToLower(strings.TrimSpace(req.ControlAction))
isAll := req.Hostname != nil && len(req.Hostname) == 1 && req.Hostname[0] == "all"
if req.Comment != "" {
if len(req.Comment) > 512 {
return false, errors.New("comment length exceeds the limit of 512 characters")
}
}
return isAll, nil
}
// hostSlotUsage holds counts of slots currently occupied on one host,
// broken down by run/suspend sub-state, mirroring HostInfo.NumRun/
// NumSSusp/NumUSusp.
type hostSlotUsage struct {
Run, SSusp, USusp int
}
// execHostsJSON mirrors the JSONB shape backend/convert_pure.go's
// buildJobInfoFromReplayJob() writes into JobInfo.ExecHosts.
type execHostsJSON struct {
HostSlots []struct {
HostName string `json:"host_name"`
NumSlots int `json:"num_slots"`
} `json:"host_slots"`
}
// computeHostSlotUsage derives per-host slot usage from the live job store,
// since store.Hosts' own NumRun/NumSSusp/NumUSusp fields are never written
// (bug-1743): a host's occupied-slot count can only be known by scanning
// which non-terminal jobs are currently placed on it.
func computeHostSlotUsage() map[string]hostSlotUsage {
usage := make(map[string]hostSlotUsage)
for _, job := range store.Jobs.List() {
if job.ExecHosts.String == "" {
continue
}
var parsed execHostsJSON
if err := json.Unmarshal([]byte(job.ExecHosts.String), &parsed); err != nil {
continue
}
for _, hs := range parsed.HostSlots {
u := usage[hs.HostName]
switch {
case job.JobStatus&models.JOBSTAT["USUSP"] != 0:
u.USusp += hs.NumSlots
case job.JobStatus&models.JOBSTAT["SSUSP"] != 0:
u.SSusp += hs.NumSlots
case job.JobStatus&models.JOBSTAT["RUN"] != 0:
u.Run += hs.NumSlots
default:
continue
}
usage[hs.HostName] = u
}
}
return usage
}
func processHostData(rows []utils.OrderMap, page_token string) utils.QueryResult {
slotUsage := computeHostSlotUsage()
for _, row := range rows {
if v, exist := row.Data["host_status"]; exist {
if v != nil {
hostStatus, _ := utils.IntOrInt64(v)
row.Data["host_status"] = models.HostStateStr(hostStatus)
}
}
// bug-1743: HostInfo.NumRun/.NumSSusp/.NumUSusp are never assigned
// anywhere else -- derive them here from the live job store so
// `hosts list` reflects actual slot usage instead of always 0.
if hostName, ok := row.Data["host_name"].(string); ok {
usage := slotUsage[hostName]
row.Data["num_run"] = usage.Run
row.Data["num_ssusp"] = usage.SSusp
row.Data["num_ususp"] = usage.USusp
}
// Process JSONB fields based on predefined list
for _, jsonbField := range models.JSONBHFields {
if v, exist := row.Data[jsonbField]; exist {
if strValue, ok := v.(string); ok {
if jsonbData, err := utils.ParseJSONB(strValue); err == nil {
row.Data[jsonbField] = jsonbData
} else {
logger.Error("Failed to parse host info JSONB field.", zap.Error(err))
}
}
}
}
//utils.FormatRowsOutput(row.Data)
}
return utils.QueryResult{
Rows: rows,
Count: len(rows),
Pagetoken: page_token,
}
}