feat(delivery): support multi-instance scheduling on a single host
- Per-host instance limit (DELIVERY_HOST_INSTANCE_LIMIT, default 4) - Optional target_host to pin a host from the candidate pool - Regenerate swagger docs
This commit is contained in:
@@ -24,6 +24,9 @@ import (
|
||||
|
||||
var dnsLabelPattern = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`)
|
||||
|
||||
// AWX inventory 主机名:允许字母数字、中划线与点(FQDN)。
|
||||
var hostNamePattern = regexp.MustCompile(`^[a-zA-Z0-9](?:[-a-zA-Z0-9.]*[a-zA-Z0-9])?$`)
|
||||
|
||||
type MySQLDeliveryInput struct {
|
||||
BusinessLineID uint64 `json:"business_line_id" binding:"required"`
|
||||
TargetID uint64 `json:"target_id" binding:"required"`
|
||||
@@ -33,9 +36,11 @@ type MySQLDeliveryInput struct {
|
||||
Topology string `json:"topology"`
|
||||
MySQLPort int `json:"mysql_port"`
|
||||
DataDisk string `json:"data_disk"`
|
||||
CPUMilli int64 `json:"cpu_milli" binding:"required"`
|
||||
MemoryMi int64 `json:"memory_mi" binding:"required"`
|
||||
StorageGi int64 `json:"storage_gi" binding:"required"`
|
||||
// 调度控制(选填):点名候选池内主机跳过自动选机,端口/配额/实机守卫照常执行
|
||||
TargetHost string `json:"target_host"`
|
||||
CPUMilli int64 `json:"cpu_milli" binding:"required"`
|
||||
MemoryMi int64 `json:"memory_mi" binding:"required"`
|
||||
StorageGi int64 `json:"storage_gi" binding:"required"`
|
||||
// 数据库配置(选填,缺省由 playbook 基线兜底)
|
||||
Timezone string `json:"timezone"`
|
||||
LowerCaseTableNames *int `json:"lower_case_table_names"`
|
||||
@@ -93,20 +98,31 @@ func parseTargetMetadata(raw string) targetMetadata {
|
||||
return meta
|
||||
}
|
||||
|
||||
// firstFreeHost 返回候选池中第一个未被占用的节点。
|
||||
func firstFreeHost(hosts []targetHost, occupied []string) *targetHost {
|
||||
taken := make(map[string]bool, len(occupied))
|
||||
for _, h := range occupied {
|
||||
taken[h] = true
|
||||
// firstFreeHost 返回候选池中非失败任务数未达单机实例上限的第一个节点;
|
||||
// limit < 1 时按 1 兜底(退化为旧的一机一实例语义)。顺序遍历天然形成"先摊开、摊满一轮再叠加"。
|
||||
func firstFreeHost(hosts []targetHost, occupied []string, limit int) *targetHost {
|
||||
if limit < 1 {
|
||||
limit = 1
|
||||
}
|
||||
for i := range hosts {
|
||||
if !taken[hosts[i].Name] {
|
||||
if hostTaskCount(occupied, hosts[i].Name) < limit {
|
||||
return &hosts[i]
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// hostTaskCount 统计某主机在占用清单(非失败任务的 target_host 列表,含重复)中的出现次数。
|
||||
func hostTaskCount(occupied []string, name string) int {
|
||||
count := 0
|
||||
for _, h := range occupied {
|
||||
if h == name {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
// 端口池 13306-13999:混合分配模型,用户留空时自动分配,可覆盖为池内指定端口。
|
||||
const (
|
||||
mysqlPortPoolStart = 13306
|
||||
@@ -344,6 +360,9 @@ func validateDeliveryInput(input MySQLDeliveryInput, dataDisks []string) error {
|
||||
return fmt.Errorf("data_disk %q is not in the allowed mount point list %v", input.DataDisk, allowed)
|
||||
}
|
||||
}
|
||||
if input.TargetHost != "" && (len(input.TargetHost) > 253 || !hostNamePattern.MatchString(input.TargetHost)) {
|
||||
return fmt.Errorf("target_host must be a valid inventory host name")
|
||||
}
|
||||
if input.Timezone != "" && !timezonePattern.MatchString(input.Timezone) {
|
||||
return fmt.Errorf("timezone must be an offset like +08:00, SYSTEM, or a named zone like Asia/Shanghai")
|
||||
}
|
||||
@@ -498,7 +517,25 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
|
||||
if err := tx.Model(&model.DeliveryTask{}).Where("target_id = ? AND target_host <> ? AND status NOT IN ?", task.TargetID, "", occupiedExclude).Pluck("target_host", &occupied).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
host := firstFreeHost(meta.Hosts, occupied)
|
||||
limit := s.cfg.DeliveryHostInstanceLimit
|
||||
var host *targetHost
|
||||
if payload.TargetHost != "" {
|
||||
// 点名主机:仅校验池内存在性与单机实例上限,端口/配额/实机守卫照常执行。
|
||||
for i := range meta.Hosts {
|
||||
if meta.Hosts[i].Name == payload.TargetHost {
|
||||
host = &meta.Hosts[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if host == nil {
|
||||
return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_host %q is not in the candidate host pool", payload.TargetHost))
|
||||
}
|
||||
if effective := max(limit, 1); hostTaskCount(occupied, host.Name) >= effective {
|
||||
return fmt.Errorf("defer: pinned host %s reached the per-host instance limit %d", host.Name, effective)
|
||||
}
|
||||
} else {
|
||||
host = firstFreeHost(meta.Hosts, occupied, limit)
|
||||
}
|
||||
if host == nil {
|
||||
return fmt.Errorf("defer: no free host available on target")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user