wonder/xinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(delivery): refine base service integrations

Browse Source
This commit is contained in:
mac
2026-07-29 16:36:27 +08:00
parent 2ca08f9360
commit 35875218c9
22 changed files with 1042 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/service/delivery.go
+188 -23
View File
@@ -129,9 +129,13 @@ type MySQLServiceLedgerItem struct {
}
type DeploymentCredentialView struct {
Username string `json:"username"`
Host string `json:"host"`
Password string `json:"password"`
Service string `json:"service"`
InstanceName string `json:"instance_name"`
Host string `json:"host"`
Port int `json:"port"`
Username string `json:"username"`
AccountHost string `json:"account_host"`
Password string `json:"password"`
}
// targetMetadata describes the native VM候选节点池以及部署形态，由 AWX inventory hosts 动态组装。
@@ -397,13 +401,19 @@ func (s *DeliveryService) getTarget(ctx context.Context, templateID uint64) (Del
return s.awxDeliveryTarget(ctx, *template)
}
func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint64, hostName string) ([]DeliveryMountPath, error) {
func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint64, hostName string, prefix string) ([]DeliveryMountPath, error) {
if targetID == 0 {
return nil, fmt.Errorf("target_id is required")
}
if hostName == "" || len(hostName) > 253 || !hostNamePattern.MatchString(hostName) {
return nil, fmt.Errorf("host must be a valid inventory host name")
}
prefix = strings.TrimSpace(prefix)
if prefix != "" {
if err := validateDirectoryLookupPrefix(prefix); err != nil {
return nil, err
}
}
template, err := s.awx.GetJobTemplate(ctx, targetID)
if err != nil {
return nil, fmt.Errorf("deployment target is unavailable: %w", err)
@@ -423,9 +433,16 @@ func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint6
return nil, fmt.Errorf("host %q is not in the deployment target inventory", hostName)
}
if s.cfg.AWXFactsTemplateID != 0 {
if err := s.refreshHostFacts(ctx, hostName); err != nil {
stdout, err := s.refreshHostFacts(ctx, hostName, prefix)
if err != nil {
return nil, err
}
if prefix != "" {
return directoryPathsFromAWXStdout(stdout), nil
}
}
if prefix != "" {
return []DeliveryMountPath{}, nil
}
facts, err := s.awx.GetHostFacts(ctx, matched.ID)
if err != nil {
@@ -438,21 +455,86 @@ func (s *DeliveryService) ListHostMountPaths(ctx context.Context, targetID uint6
return items, nil
}
func (s *DeliveryService) refreshHostFacts(ctx context.Context, hostName string) error {
func validateDirectoryLookupPrefix(prefix string) error {
if prefix == "" {
return nil
}
if len(prefix) > 512 {
return fmt.Errorf("path prefix is too long")
}
if !strings.HasPrefix(prefix, "/") {
return fmt.Errorf("path prefix must be an absolute path")
}
if strings.ContainsRune(prefix, 0) {
return fmt.Errorf("path prefix contains invalid characters")
}
return nil
}
func directoryPathsFromAWXStdout(stdout string) []DeliveryMountPath {
const marker = "XINFRA_PATH_COMPLETIONS_JSON="
for _, line := range strings.Split(stdout, "\n") {
line = strings.TrimSpace(line)
idx := strings.Index(line, marker)
if idx < 0 {
continue
}
if items, ok := parseDirectoryCompletionJSON(line[idx+len(marker):]); ok {
sort.Slice(items, func(i, j int) bool {
return items[i].Path < items[j].Path
})
return items
}
}
return []DeliveryMountPath{}
}
func parseDirectoryCompletionJSON(raw string) ([]DeliveryMountPath, bool) {
raw = strings.TrimSpace(raw)
candidates := []string{raw}
if strings.Contains(raw, `\"`) {
candidates = append(candidates, strings.ReplaceAll(raw, `\"`, `"`))
}
for _, candidate := range candidates {
start := strings.Index(candidate, "[")
end := strings.LastIndex(candidate, "]")
if start < 0 || end < start {
continue
}
var items []DeliveryMountPath
if err := json.Unmarshal([]byte(candidate[start:end+1]), &items); err == nil {
return items, true
}
}
return nil, false
}
func (s *DeliveryService) refreshHostFacts(ctx context.Context, hostName string, lookupPath string) (string, error) {
extraVars := map[string]any{
"target_hosts": hostName,
}
if lookupPath != "" {
extraVars["lookup_path"] = lookupPath
}
job, err := s.awx.Launch(ctx, s.cfg.AWXFactsTemplateID, AWXLaunchRequest{
Limit: hostName,
Limit: hostName,
ExtraVars: extraVars,
})
if err != nil {
return fmt.Errorf("launch AWX facts job: %w", err)
return "", fmt.Errorf("launch AWX facts job: %w", err)
}
done, err := s.awx.WaitJob(ctx, strconv.FormatUint(job.ID, 10), time.Duration(s.cfg.AWXFactsTimeoutSeconds)*time.Second)
if err != nil {
return err
return "", err
}
if done.Status != "successful" || done.Failed {
return fmt.Errorf("AWX facts job %d finished with status %s", done.ID, done.Status)
return "", fmt.Errorf("AWX facts job %d finished with status %s", done.ID, done.Status)
}
return nil
stdout, err := s.awx.JobStdout(ctx, strconv.FormatUint(job.ID, 10))
if err != nil {
return "", err
}
return stdout, nil
}
func (s *DeliveryService) awxDeliveryTarget(ctx context.Context, template AWXJobTemplate) (DeliveryTarget, error) {
@@ -503,6 +585,9 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin
if len(credentialInput["root@localhost"]) < 16 || len(credentialInput["xinfra_admin@%"]) < 16 {
return nil, false, fmt.Errorf("mysql passwords must be at least 16 characters")
}
if !mysqlPasswordPattern.MatchString(credentialInput["root@localhost"]) || !mysqlPasswordPattern.MatchString(credentialInput["xinfra_admin@%"]) {
return nil, false, fmt.Errorf("mysql passwords may only contain letters and digits")
}
}
input.MySQLRootPassword = ""
input.MySQLAdminPassword = ""
@@ -657,6 +742,8 @@ var (
// timezone 仅接受偏移量（±HH:MM）、SYSTEM 或命名时区（如 Asia/Shanghai）。
var timezonePattern = regexp.MustCompile(`^([+-](0\d|1[0-4]):[0-5]\d|SYSTEM|[A-Za-z]+(?:/[A-Za-z0-9_+-]+)+)$`)
var mysqlPasswordPattern = regexp.MustCompile(`^[A-Za-z0-9]+$`)
func validateDeliveryInput(input MySQLDeliveryInput, _ []string) error {
if len(input.Namespace) > 63 || !dnsLabelPattern.MatchString(input.Namespace) {
return fmt.Errorf("namespace must be a valid Kubernetes DNS label")
@@ -773,7 +860,13 @@ func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin
query = query.Where("business_line_id IN (?)", s.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID))
}
var tasks []model.DeliveryTask
return tasks, query.Find(&tasks).Error
if err := query.Find(&tasks).Error; err != nil {
return nil, err
}
if err := s.fillCredentialAvailability(ctx, tasks); err != nil {
return nil, err
}
return tasks, nil
}
func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]MySQLServiceLedgerItem, error) {
@@ -916,6 +1009,11 @@ func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uin
if err := query.First(&task).Error; err != nil {
return nil, nil, err
}
taskItems := []model.DeliveryTask{task}
if err := s.fillCredentialAvailability(ctx, taskItems); err != nil {
return nil, nil, err
}
task = taskItems[0]
var events []model.TaskEvent
if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).Order("id ASC").Find(&events).Error; err != nil {
return nil, nil, err
@@ -923,6 +1021,41 @@ func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uin
return &task, events, nil
}
func (s *DeliveryService) fillCredentialAvailability(ctx context.Context, tasks []model.DeliveryTask) error {
if len(tasks) == 0 {
return nil
}
taskIDs := make([]string, 0, len(tasks))
for i := range tasks {
if tasks[i].Status == model.TaskFinished || tasks[i].Status == model.TaskRegisterFailed {
taskIDs = append(taskIDs, tasks[i].ID)
}
}
if len(taskIDs) == 0 {
return nil
}
var rows []struct {
TaskID string
Count int64
}
if err := s.db.WithContext(ctx).
Model(&model.DeploymentCredential{}).
Select("task_id, count(*) as count").
Where("task_id IN ? AND status = ?", taskIDs, "available").
Group("task_id").
Scan(&rows).Error; err != nil {
return err
}
available := make(map[string]bool, len(rows))
for _, row := range rows {
available[row.TaskID] = row.Count > 0
}
for i := range tasks {
tasks[i].CredentialAvailable = available[tasks[i].ID]
}
return nil
}
func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskID string, userID uint64, isAdmin bool) ([]DeploymentCredentialView, error) {
task, _, err := s.GetTask(ctx, taskID, userID, isAdmin)
if err != nil {
@@ -933,6 +1066,10 @@ func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskI
}
var out []DeploymentCredentialView
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var result model.DeploymentResult
if err := tx.Where("task_id = ?", taskID).First(&result).Error; err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
var credentials []model.DeploymentCredential
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
Where("task_id = ? AND status = ?", taskID, "available").
@@ -949,22 +1086,20 @@ func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskI
return err
}
out = append(out, DeploymentCredentialView{
Username: credential.Username,
Host: credential.AccountHost,
Password: password,
Service: firstNonEmptyDeliveryValue(result.Component, credential.Component, task.Component, "mysql"),
InstanceName: firstNonEmptyDeliveryValue(result.InstanceName, credential.InstanceName, task.InstanceName),
Host: firstNonEmptyDeliveryValue(result.Host, task.TargetHostIP),
Port: firstNonZero(result.Port, task.MySQLPort),
Username: credential.Username,
AccountHost: credential.AccountHost,
Password: password,
})
}
ids := make([]uint64, 0, len(credentials))
for _, credential := range credentials {
ids = append(ids, credential.ID)
}
now := time.Now()
return tx.Model(&model.DeploymentCredential{}).Where("id IN ?", ids).Updates(map[string]any{
"status": "viewed",
"viewed_by": userID,
"viewed_at": now,
"updated_at": now,
}).Error
return tx.Unscoped().Where("id IN ?", ids).Delete(&model.DeploymentCredential{}).Error
})
if err != nil {
return nil, err
@@ -972,6 +1107,24 @@ func (s *DeliveryService) RevealDeploymentCredentials(ctx context.Context, taskI
return out, nil
}
func firstNonEmptyDeliveryValue(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return strings.TrimSpace(value)
}
}
return ""
}
func firstNonZero(values ...int) int {
for _, value := range values {
if value != 0 {
return value
}
}
return 0
}
func (s *DeliveryService) SubscribeTask(taskID string) (<-chan DeliveryTaskSnapshot, func()) {
ch := make(chan DeliveryTaskSnapshot, 8)
s.streamMu.Lock()
@@ -1170,7 +1323,7 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT
return s.failInTransaction(tx, &task, model.TaskValidationFailed, "deployment target has no candidate hosts")
}
var occupied []string
occupiedExclude := []string{model.TaskExecutionFailed, model.TaskValidationFailed, model.TaskCanceled}
occupiedExclude := allocationReleasedTaskStatuses()
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
}
@@ -1641,6 +1794,18 @@ func terminalTaskStatuses() []string {
}
}
func allocationReleasedTaskStatuses() []string {
return []string{
model.TaskFinished,
model.TaskExecutionFailed,
model.TaskValidationFailed,
model.TaskRegisterFailed,
model.TaskCanceled,
model.TaskRolledBack,
model.TaskRollbackAck,
}
}
func awxNotificationMessage(input AWXJobNotificationInput) string {
status := strings.TrimSpace(input.Status)
name := strings.TrimSpace(input.Name)