修改选择host的逻辑，先保证单主机单实例的发布流程

Author: aice030

internal/deploy/database/instance_repo.go

@@ -351,3 +351,42 @@ func (r *InstanceRepo) GetExistingInstancePorts(serviceName, instanceIP string)
 
 	return ports, nil
 }
+
+// GetHostsRunningServiceVersion 查询已运行指定服务指定版本的主机ID列表
+func (r *InstanceRepo) GetHostsRunningServiceVersion(serviceName, serviceVersion string) ([]string, error) {
+	// 参数验证
+	if serviceName == "" {
+		return nil, fmt.Errorf("serviceName cannot be empty")
+	}
+	if serviceVersion == "" {
+		return nil, fmt.Errorf("serviceVersion cannot be empty")
+	}
+
+	query := `
+		SELECT DISTINCT host_id 
+		FROM instances 
+		WHERE service_name = $1 AND service_version = $2 AND status = 'active'
+		ORDER BY host_id
+	`
+
+	rows, err := r.db.Query(query, serviceName, serviceVersion)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query hosts running service %s version %s: %w", serviceName, serviceVersion, err)
+	}
+	defer rows.Close()
+
+	var hostIDs []string
+	for rows.Next() {
+		var hostID string
+		if err := rows.Scan(&hostID); err != nil {
+			return nil, fmt.Errorf("failed to scan host_id: %w", err)
+		}
+		hostIDs = append(hostIDs, hostID)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("error iterating host IDs: %w", err)
+	}
+
+	return hostIDs, nil
+}

internal/deploy/service/deploy_service.go

@@ -635,7 +635,7 @@ func (f *floyDeployService) deployToSingleInstance(instanceIP, service, version,
 
 	// 3. 推送配置文件
 	if wantConfig {
-		if err := f.pushConfig(instanceIP, service, fversion); err != nil {
+		if err := f.pushConfig(instanceIP, service, fversion, version, packageFilePath); err != nil {
 			return fmt.Errorf("推送配置文件失败: %v", err)
 		}
 	}
@@ -667,14 +667,14 @@ func (f *floyDeployService) rollbackToSingleInstance(instanceIP, service, target
 			return fmt.Errorf("推送回滚包文件失败: %v", err)
 		}
 	}
-	// // 2. 推送回滚包文件
+	// 2. 推送回滚包文件
 	// if err := f.pushPackage(instanceIP, service, fversion, targetVersion, packageFilePath, md5sum); err != nil {
 	// 	return fmt.Errorf("推送回滚包文件失败: %v", err)
 	// }
 
 	// 3. 推送配置文件
 	if wantConfig {
-		if err := f.pushConfig(instanceIP, service, fversion); err != nil {
+		if err := f.pushConfig(instanceIP, service, fversion, targetVersion, packageFilePath); err != nil {
 			return fmt.Errorf("推送配置文件失败: %v", err)
 		}
 	}
@@ -859,13 +859,15 @@ func (f *floyDeployService) pushPackage(instanceIP, service, fversion, version s
 }
 
 // pushConfig 推送配置文件
-func (f *floyDeployService) pushConfig(instanceIP, service, fversion string) error {
+func (f *floyDeployService) pushConfig(instanceIP, service, fversion, version, packageFilePath string) error {
 	baseURL := fmt.Sprintf("http://%s:%s", instanceIP, f.port)
 
-	// 简单的配置文件示例
-	configContent := fmt.Sprintf("# Configuration for %s\nservice.name=%s\nservice.version=%s\n",
-		service, service, fversion)
-	configMD5 := md5.Sum([]byte(configContent))
+	// 从修改后的包文件中提取配置文件内容
+	configContent, err := f.extractConfigFromPackage(packageFilePath)
+	if err != nil {
+		return fmt.Errorf("failed to extract config from package: %v", err)
+	}
+	configMD5 := md5.Sum(configContent)
 
 	// 使用 multipart.Writer 构造请求体
 	var buf bytes.Buffer
@@ -879,7 +881,7 @@ func (f *floyDeployService) pushConfig(instanceIP, service, fversion string) err
 
 	// 创建文件字段，设置 Content-Md5 头
 	header := make(map[string][]string)
-	header["Content-Disposition"] = []string{`form-data; name="file"; filename="app.conf"`}
+	header["Content-Disposition"] = []string{fmt.Sprintf(`form-data; name="file"; filename="%s/config.yaml"`, version)}
 	header["Content-Type"] = []string{"application/octet-stream"}
 	header["Content-Md5"] = []string{base64.URLEncoding.EncodeToString(configMD5[:])}
 	header["File-Mode"] = []string{"644"}
@@ -890,7 +892,7 @@ func (f *floyDeployService) pushConfig(instanceIP, service, fversion string) err
 	}
 
 	// 写入配置文件内容
-	_, err = fileWriter.Write([]byte(configContent))
+	_, err = fileWriter.Write(configContent)
 	if err != nil {
 		return fmt.Errorf("failed to write config data: %v", err)
 	}
@@ -1545,3 +1547,28 @@ func (f *floyDeployService) createTarGz(src, dest string) error {
 		return nil
 	})
 }
+
+// extractConfigFromPackage 从包文件中提取配置文件内容
+func (f *floyDeployService) extractConfigFromPackage(packageFilePath string) ([]byte, error) {
+	// 创建临时目录
+	tempDir, err := os.MkdirTemp("", "extract-config-*")
+	if err != nil {
+		return nil, fmt.Errorf("创建临时目录失败: %v", err)
+	}
+	defer os.RemoveAll(tempDir)
+
+	// 解压包文件到临时目录
+	err = f.extractTarGz(packageFilePath, tempDir)
+	if err != nil {
+		return nil, fmt.Errorf("解压包文件失败: %v", err)
+	}
+
+	// 读取配置文件
+	configPath := filepath.Join(tempDir, "config.yaml")
+	configContent, err := os.ReadFile(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("读取配置文件失败: %v", err)
+	}
+
+	return configContent, nil
+}

internal/deploy/service/internal_utils.go

@@ -156,15 +156,53 @@ func CheckHostHealth(hostIpAddress string) (bool, error) {
 }
 
 // SelectHostForNewInstance 为新实例选择合适的主机
+// 查询数据库，选择未运行当前服务当前版本的主机
 func SelectHostForNewInstance(availableHosts []*model.HostInfo, service string, version string) (*model.HostInfo, error) {
 	// 参数验证
 	if len(availableHosts) == 0 {
 		return nil, fmt.Errorf("no available hosts")
 	}
+	if service == "" {
+		return nil, fmt.Errorf("service name cannot be empty")
+	}
+	if version == "" {
+		return nil, fmt.Errorf("service version cannot be empty")
+	}
+
+	// 获取数据库连接
+	_, err := initDatabase()
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize database: %w", err)
+	}
+
+	// 查询已运行当前服务当前版本的主机ID列表
+	runningHostIDs, err := instanceRepo.GetHostsRunningServiceVersion(service, version)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query hosts running service %s version %s: %w", service, version, err)
+	}
+
+	// 创建已运行主机ID的映射，便于快速查找
+	runningHostMap := make(map[string]bool)
+	for _, hostID := range runningHostIDs {
+		runningHostMap[hostID] = true
+	}
+
+	// 筛选出未运行当前服务当前版本的主机
+	availableHostsForService := []*model.HostInfo{}
+	for _, host := range availableHosts {
+		if !runningHostMap[host.HostID] {
+			availableHostsForService = append(availableHostsForService, host)
+		}
+	}
+
+	// 如果没有可用的主机，返回错误
+	if len(availableHostsForService) == 0 {
+		return nil, fmt.Errorf("all hosts are already running service %s version %s", service, version)
+	}
 
-	// 随机选择一个主机
-	randomIndex := rand.Intn(len(availableHosts))
-	selectedHost := availableHosts[randomIndex]
+	// 从可用主机中随机选择一个
+	randomIndex := rand.Intn(len(availableHostsForService))
+	selectedHost := availableHostsForService[randomIndex]
 
 	return selectedHost, nil
 }