OCDP/ocdp-go
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: complete E2E deployment flow with storage layered config and values template versioning

Browse Source 
- Instance deployment: charts browser, deploy modal, instances list
- Values Template version management (create/history/rollback)
- Storage layered config (cluster > workspace > shared priority)
- Cluster credential decryptIfNeeded for mixed encrypted/plaintext kubeconfig
- YAML syntax validation (client-side + server-side warning)
- Frontend: charts, instances, storage, templates, admin pages
- Backend: storage service, instance service, cluster service, helm client
- Multi-Tenant Kubeconfig.md: added by user
This commit is contained in:
Ivan087
2026-04-30 16:31:00 +08:00
parent 985369d40f
commit 47849042a7
42 changed files with 2029 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/internal/adapter/input/http/dto/converter.go
View File

@ -76,6 +76,7 @@ func WorkspaceDTOFromEntity(workspace *entity.Workspace) *WorkspaceDTO {
return &WorkspaceDTO{
ID: workspace.ID,
Name: workspace.Name,
ClusterIDs: workspace.ClusterIDs,
Description: workspace.Description,
CreatedBy: workspace.CreatedBy,
CreatedAt: workspace.CreatedAt,

25
backend/internal/adapter/input/http/dto/storage_dto.go
View File

@ -7,6 +7,7 @@ type CreateStorageRequest struct {
Description string `json:"description"`
IsDefault bool `json:"is_default"`
IsShared bool `json:"is_shared"`
ClusterID string `json:"cluster_id,omitempty"` // 用于 cluster-level storage
// NFS 配置
NFS NFSConfigDTO `json:"nfs,omitempty"`
@ -23,6 +24,7 @@ type UpdateStorageRequest struct {
Description string `json:"description"`
IsDefault bool `json:"is_default"`
IsShared bool `json:"is_shared"`
ClusterID string `json:"cluster_id,omitempty"` // 用于 cluster-level storage
// NFS 配置
NFS NFSConfigDTO `json:"nfs,omitempty"`
@ -52,17 +54,18 @@ type HostPathConfigDTO struct {
// StorageResponse 存储后端响应
type StorageResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id,omitempty"`
OwnerID string `json:"owner_id,omitempty"`
Name string `json:"name"`
Type string `json:"type"`
Config StorageConfigDTO `json:"config"`
Description string `json:"description"`
IsDefault bool `json:"is_default"`
IsShared bool `json:"is_shared"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
ID string `json:"id"`
WorkspaceID string `json:"workspace_id,omitempty"`
ClusterID string `json:"cluster_id,omitempty"`
OwnerID string `json:"owner_id,omitempty"`
Name string `json:"name"`
Type string `json:"type"`
Config StorageConfigDTO `json:"config"`
Description string `json:"description"`
IsDefault bool `json:"is_default"`
IsShared bool `json:"is_shared"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
}
// StorageConfigDTO 存储配置（脱敏后）

26
backend/internal/adapter/input/http/dto/workspace_dto.go
View File

@ -4,24 +4,32 @@ import "time"
// WorkspaceDTO 工作空间 DTO
type WorkspaceDTO struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID string `json:"id"`
Name string `json:"name"`
ClusterIDs []string `json:"cluster_ids,omitempty"`
Quotas []*QuotaDTO `json:"quotas,omitempty"`
Description string `json:"description,omitempty"`
CreatedBy string `json:"created_by"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// CreateWorkspaceRequest 创建工作空间请求
// CreateWorkspaceRequest 创建工作空间请求（包含配额设置）
type CreateWorkspaceRequest struct {
Name string `json:"name" validate:"required"`
Description string `json:"description"`
ClusterIDs []string `json:"cluster_ids"`
// Quotas can be set during creation
CPU *QuotaValue `json:"cpu"`
GPU *QuotaValue `json:"gpu"`
GPUMemory *QuotaValue `json:"gpu_memory"`
}
// UpdateWorkspaceRequest 更新工作空间请求
type UpdateWorkspaceRequest struct {
Name string `json:"name"`
Description string `json:"description"`
Name string `json:"name"`
Description string `json:"description"`
ClusterIDs []string `json:"cluster_ids"`
}
// QuotaDTO 配额 DTO

8
backend/internal/adapter/input/http/rest/auth_handler.go
View File

@ -83,14 +83,14 @@ func (h *AuthHandler) Login(w http.ResponseWriter, r *http.Request) {
// 获取用户信息
// TODO: 从 token 解析用户信息或从服务获取
// 返回响应
// 返回响应 - 使用 respondSuccess 包装，与其他 API 保持一致
response := &dto.AuthResponse{
AccessToken: accessToken,
RefreshToken: refreshToken,
Username: req.Username,
}
respondJSON(w, http.StatusOK, response)
respondSuccess(w, "Login successful", response)
}
// RefreshToken 刷新 Token
@ -117,11 +117,11 @@ func (h *AuthHandler) RefreshToken(w http.ResponseWriter, r *http.Request) {
return
}
// 返回响应
// 返回响应 - 使用 respondSuccess 包装
response := &dto.AuthResponse{
AccessToken: newAccessToken,
RefreshToken: req.RefreshToken,
}
respondJSON(w, http.StatusOK, response)
respondSuccess(w, "Token refreshed", response)
}

1
backend/internal/adapter/input/http/rest/instance_handler.go
View File

@ -188,6 +188,7 @@ func (h *InstanceHandler) ListInstances(w http.ResponseWriter, r *http.Request)
LastOperation: string(instance.LastOperation),
LastError: instance.LastError,
Revision: instance.Revision,
Values: instance.Values,
CreatedAt: instance.CreatedAt.Format("2006-01-02T15:04:05Z07:00"),
UpdatedAt: instance.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
})

49
backend/internal/adapter/input/http/rest/storage_handler.go
View File

@ -10,6 +10,14 @@ import (
"github.com/ocdp/cluster-service/internal/domain/service"
)
// StorageResolutionResponse 分层存储解析响应
type StorageResolutionResponse struct {
Storage *dto.StorageResponse `json:"storage,omitempty"`
ValuesYAML string `json:"values_yaml,omitempty"`
Source string `json:"source,omitempty"` // workspace, cluster, shared
Message string `json:"message,omitempty"`
}
// StorageHandler Storage Backend Handler
type StorageHandler struct {
storageService *service.StorageService
@ -77,6 +85,7 @@ func (h *StorageHandler) CreateStorage(w http.ResponseWriter, r *http.Request) {
req.Description,
req.IsDefault,
req.IsShared,
req.ClusterID,
)
if err != nil {
respondError(w, http.StatusBadRequest, "Failed to create storage backend", err.Error())
@ -252,6 +261,45 @@ func (h *StorageHandler) DeleteStorage(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}
// ResolveStorage 预览分层存储解析结果
// @Summary 预览分层存储解析结果
// @Description 根据 cluster_id 和 workspace_id 解析出最终生效的存储配置
// @Tags Storage
// @Produce json
// @Security BearerAuth
// @Param cluster_id query string false "Cluster ID"
// @Param workspace_id query string false "Workspace ID"
// @Success 200 {object} StorageResolutionResponse
// @Router /storage-backends/resolve [get]
func (h *StorageHandler) ResolveStorage(w http.ResponseWriter, r *http.Request) {
clusterID := r.URL.Query().Get("cluster_id")
workspaceID := r.URL.Query().Get("workspace_id")
if workspaceID == "" {
workspaceID = r.Header.Get("X-Workspace-ID")
}
resolution, err := h.storageService.ResolveStorageConfig(r.Context(), clusterID, workspaceID)
if err != nil {
respondError(w, http.StatusInternalServerError, "Failed to resolve storage config", err.Error())
return
}
if resolution == nil || resolution.Storage == nil {
respondJSON(w, http.StatusOK, &StorageResolutionResponse{
Message: "No default storage configured",
})
return
}
response := &StorageResolutionResponse{
Storage: toStorageResponse(resolution.Storage),
ValuesYAML: resolution.ValuesYAML,
Source: resolution.Source,
}
respondJSON(w, http.StatusOK, response)
}
// toStorageResponse 转换为响应 DTO
func toStorageResponse(storage *entity.StorageBackend) *dto.StorageResponse {
config := dto.StorageConfigDTO{}
@ -278,6 +326,7 @@ func toStorageResponse(storage *entity.StorageBackend) *dto.StorageResponse {
return &dto.StorageResponse{
ID: storage.ID,
WorkspaceID: storage.WorkspaceID,
ClusterID: storage.ClusterID,
OwnerID: storage.OwnerID,
Name: storage.Name,
Type: string(storage.Type),

31
backend/internal/adapter/input/http/rest/workspace_handler.go
View File

@ -26,7 +26,7 @@ func NewWorkspaceHandler(workspaceService *service.WorkspaceService, authService
// CreateWorkspace 创建工作空间
// @Summary 创建工作空间
// @Description 创建新的工作空间（Admin 专用）
// @Description 创建新的工作空间（Admin 专用，支持 cluster_ids 和初始配额）
// @Tags workspace
// @Accept json
// @Produce json
@ -48,7 +48,31 @@ func (h *WorkspaceHandler) CreateWorkspace(w http.ResponseWriter, r *http.Reques
// 获取创建者 ID
userID := GetUserIDFromRequest(r)
workspace, err := h.workspaceService.Create(r.Context(), req.Name, req.Description, userID)
// 准备配额
quotas := make(map[entity.ResourceType]struct {
HardLimit float64
SoftLimit float64
})
if req.CPU != nil {
quotas[entity.ResourceCPU] = struct {
HardLimit float64
SoftLimit float64
}{req.CPU.HardLimit, req.CPU.SoftLimit}
}
if req.GPU != nil {
quotas[entity.ResourceGPU] = struct {
HardLimit float64
SoftLimit float64
}{req.GPU.HardLimit, req.GPU.SoftLimit}
}
if req.GPUMemory != nil {
quotas[entity.ResourceGPUMemory] = struct {
HardLimit float64
SoftLimit float64
}{req.GPUMemory.HardLimit, req.GPUMemory.SoftLimit}
}
workspace, err := h.workspaceService.Create(r.Context(), req.Name, req.Description, userID, req.ClusterIDs, quotas)
if err != nil {
respondError(w, http.StatusBadRequest, err.Error(), "")
return
@ -129,6 +153,9 @@ func (h *WorkspaceHandler) UpdateWorkspace(w http.ResponseWriter, r *http.Reques
if req.Description != "" {
workspace.Description = req.Description
}
if req.ClusterIDs != nil {
workspace.ClusterIDs = req.ClusterIDs
}
if err := h.workspaceService.Update(r.Context(), workspace); err != nil {
respondError(w, http.StatusBadRequest, err.Error(), "")

9
backend/internal/adapter/output/helm/real/helm_client.go
View File

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"log"
"os"
"path/filepath"
"time"
@ -113,22 +114,24 @@ func (h *HelmClient) Install(ctx context.Context, cluster *entity.Cluster, insta
install.Namespace = instance.Namespace
install.CreateNamespace = true
install.Wait = true
install.Timeout = 5 * time.Minute
install.Timeout = 1 * time.Minute
// 加载 Chart（从本地路径或 OCI registry）
// 这里简化处理，假设 chart 已经被拉取到本地
chartPath := fmt.Sprintf("/tmp/charts/%s-%s.tgz", instance.Chart, instance.Version)
chart, err := loader.Load(chartPath)
if err != nil {
return fmt.Errorf("failed to load chart: %w", err)
}
// 执行安装
log.Printf("[helm-install] step=run instance=%s values=%v", instance.Name, instance.Values)
t0 := time.Now()
rel, err := install.Run(chart, instance.Values)
log.Printf("[helm-install] step=runDone instance=%s elapsed=%v err=%v", instance.Name, time.Since(t0), err)
if err != nil {
return fmt.Errorf("failed to install release: %w", err)
}
log.Printf("[helm-install] step=done instance=%s revision=%d", instance.Name, rel.Version)
// 更新 revision（状态由调用方根据操作结果设置）
instance.Revision = rel.Version

143
backend/internal/adapter/output/oci/real/oci_client.go
View File

@ -7,6 +7,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"path/filepath"
@ -74,23 +75,147 @@ func (c *OCIClient) getRegistry(reg *entity.Registry) (*remote.Registry, error)
}
// ListRepositories 列出 Registry 中的所有 repositories
// 优先使用 OCI _catalog API，失败时回退到 Harbor REST API v2
func (c *OCIClient) ListRepositories(ctx context.Context, registry *entity.Registry) ([]string, error) {
reg, err := c.getRegistry(registry)
if err != nil {
return nil, err
}
repositories := make([]string, 0)
err = reg.Repositories(ctx, "", func(repos []string) error {
repositories = append(repositories, repos...)
return nil
})
// 尝试 OCI _catalog API
reg, err := c.getRegistry(registry)
log.Printf("[DEBUG ListRepositories] registry=%s, getRegistry err=%v", registry.URL, err)
if err == nil {
err = reg.Repositories(ctx, "", func(repos []string) error {
log.Printf("[DEBUG ListRepositories] OCI got repos batch: %d", len(repos))
repositories = append(repositories, repos...)
return nil
})
log.Printf("[DEBUG ListRepositories] OCI reg.Repositories returned: err=%v, total_repos=%d", err, len(repositories))
}
log.Printf("[DEBUG ListRepositories] post-OCI check: err=%v, repos_count=%d", err, len(repositories))
if err == nil && len(repositories) > 0 {
log.Printf("[DEBUG ListRepositories] OCI success, returning %d repos", len(repositories))
return repositories, nil
}
// 回退: 使用 Harbor REST API v2
log.Printf("[Harbor Fallback] OCI failed (err=%v, repos=%d), checking if Harbor...", err, len(repositories))
log.Printf("[Harbor Fallback] registry.URL=%s, contains 'harbor'=%v", registry.URL, strings.Contains(registry.URL, "harbor"))
if strings.Contains(registry.URL, "harbor") {
log.Printf("[Harbor Fallback] Yes, this is Harbor! Calling Harbor REST API...")
repos, fallbackErr := c.listHarborRepositories(registry)
log.Printf("[Harbor Fallback] Got %d repos, err=%v", len(repos), fallbackErr)
if fallbackErr == nil && len(repos) > 0 {
log.Printf("[Harbor Fallback] Returning %d repos from Harbor API", len(repos))
return repos, nil
}
if err != nil {
return nil, fmt.Errorf("failed to list repositories: %w", err)
}
return nil, fallbackErr
}
if err != nil {
return nil, fmt.Errorf("failed to list repositories: %w", err)
}
return repositories, nil
}
// listHarborRepositories 使用 Harbor REST API v2 获取仓库列表
func (c *OCIClient) listHarborRepositories(registry *entity.Registry) ([]string, error) {
// 解析 Harbor URL 基础地址
baseURL := registry.URL
baseURL = strings.TrimSuffix(baseURL, "/")
baseURL = strings.TrimPrefix(baseURL, "https://")
baseURL = strings.TrimPrefix(baseURL, "http://")
harborHost := "https://" + baseURL
// 获取认证信息
username := registry.Username
password := registry.Password
if username == "" || password == "" {
username = os.Getenv("HARBOR_USERNAME")
password = os.Getenv("HARBOR_PASSWORD")
}
// 获取项目列表
projectsURL := harborHost + "/api/v2.0/projects"
req, err := http.NewRequest("GET", projectsURL, nil)
if err != nil {
return nil, err
}
req.SetBasicAuth(username, password)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("failed to list projects: status %d", resp.StatusCode)
}
var projects []struct {
Name string `json:"name"`
}
if err := json.NewDecoder(resp.Body).Decode(&projects); err != nil {
return nil, err
}
repositories := make([]string, 0)
pageSize := 100
for _, project := range projects {
page := 1
log.Printf("[listHarborRepositories] Processing project: %s", project.Name)
for {
reposURL := fmt.Sprintf("%s/api/v2.0/projects/%s/repositories?page=%d&page_size=%d",
harborHost, project.Name, page, pageSize)
req, err := http.NewRequest("GET", reposURL, nil)
if err != nil {
log.Printf("[listHarborRepositories] page %d: NewRequest error: %v", page, err)
break
}
req.SetBasicAuth(username, password)
resp, err := c.httpClient.Do(req)
if err != nil {
log.Printf("[listHarborRepositories] page %d: Do error: %v", page, err)
break
}
if resp.StatusCode != http.StatusOK {
bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 200))
resp.Body.Close()
log.Printf("[listHarborRepositories] page %d: HTTP %d, body: %s", page, resp.StatusCode, string(bodyBytes))
break
}
var repos []struct {
Name string `json:"name"`
}
if err := json.NewDecoder(resp.Body).Decode(&repos); err != nil {
resp.Body.Close()
log.Printf("[listHarborRepositories] page %d: Decode error: %v", page, err)
break
}
resp.Body.Close()
log.Printf("[listHarborRepositories] page %d: got %d repos", page, len(repos))
if len(repos) == 0 {
break
}
for _, repo := range repos {
repositories = append(repositories, repo.Name)
}
page++
}
}
log.Printf("[listHarborRepositories] Total repos collected: %d", len(repositories))
return repositories, nil
}

23
backend/internal/adapter/output/persistence/mock/storage_repository_mock.go
View File

@ -83,7 +83,28 @@ func (r *StorageRepositoryMock) GetDefault(ctx context.Context, workspaceID stri
return s, nil
}
}
return nil, entity.ErrStorageNotFound
return nil, nil
}
// GetByCluster 获取 cluster 关联的存储后端
func (r *StorageRepositoryMock) GetByCluster(ctx context.Context, clusterID string) ([]*entity.StorageBackend, error) {
var result []*entity.StorageBackend
for _, s := range r.storages {
if s.ClusterID == clusterID {
result = append(result, s)
}
}
return result, nil
}
// GetDefaultByCluster 获取 cluster 的默认存储后端
func (r *StorageRepositoryMock) GetDefaultByCluster(ctx context.Context, clusterID string) (*entity.StorageBackend, error) {
for _, s := range r.storages {
if s.ClusterID == clusterID && s.IsDefault {
return s, nil
}
}
return nil, nil
}
// List 列出所有存储（管理员用）

51
backend/internal/adapter/output/persistence/postgres/cluster_repository.go
View File

@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"fmt"
"log"
"time"
"github.com/google/uuid"
@ -124,11 +125,11 @@ func (r *ClusterRepository) GetByID(ctx context.Context, id string) (*entity.Clu
return nil, fmt.Errorf("failed to get cluster: %w", err)
}
// 解密敏感数据
cluster.CAData, _ = r.encryptor.Decrypt(encryptedCAData)
cluster.CertData, _ = r.encryptor.Decrypt(encryptedCertData)
cluster.KeyData, _ = r.encryptor.Decrypt(encryptedKeyData)
cluster.Token, _ = r.encryptor.Decrypt(encryptedToken)
// 解密敏感数据（检测 kubeconfig 格式则跳过解密）
cluster.CAData = r.decryptIfNeeded(encryptedCAData, "ca_data")
cluster.CertData = r.decryptIfNeeded(encryptedCertData, "cert_data")
cluster.KeyData = r.decryptIfNeeded(encryptedKeyData, "key_data")
cluster.Token = r.decryptIfNeeded(encryptedToken, "token")
return cluster, nil
}
@ -169,15 +170,35 @@ func (r *ClusterRepository) GetByName(ctx context.Context, name string) (*entity
return nil, fmt.Errorf("failed to get cluster: %w", err)
}
// 解密敏感数据
cluster.CAData, _ = r.encryptor.Decrypt(encryptedCAData)
cluster.CertData, _ = r.encryptor.Decrypt(encryptedCertData)
cluster.KeyData, _ = r.encryptor.Decrypt(encryptedKeyData)
cluster.Token, _ = r.encryptor.Decrypt(encryptedToken)
// 解密敏感数据（检测 kubeconfig 格式则跳过解密）
cluster.CAData = r.decryptIfNeeded(encryptedCAData, "ca_data")
cluster.CertData = r.decryptIfNeeded(encryptedCertData, "cert_data")
cluster.KeyData = r.decryptIfNeeded(encryptedKeyData, "key_data")
cluster.Token = r.decryptIfNeeded(encryptedToken, "token")
return cluster, nil
}
// decryptIfNeeded 解密数据。如果数据以 "apiVersion:" 或 "kind:" 开头（kubeconfig 格式），
// 则跳过解密直接返回原值。
func (r *ClusterRepository) decryptIfNeeded(data string, fieldName string) string {
if data == "" {
return ""
}
// 检测 kubeconfig 格式（明文 YAML）
if (len(data) > 10 && data[:11] == "apiVersion:") ||
(len(data) > 5 && data[:5] == "kind:") {
return data
}
// 否则尝试解密
decrypted, err := r.encryptor.Decrypt(data)
if err != nil {
log.Printf("[ClusterRepository] WARNING: failed to decrypt %s for field %s: %v (field will be empty)", data[:min(50, len(data))], fieldName, err)
return ""
}
return decrypted
}
// Update 更新集群
func (r *ClusterRepository) Update(ctx context.Context, cluster *entity.Cluster) error {
cluster.UpdatedAt = time.Now()
@ -352,18 +373,18 @@ func (r *ClusterRepository) scanClusters(rows *sql.Rows) ([]*entity.Cluster, err
cluster.OwnerID = ownerID.String
cluster.DefaultNamespace = defaultNamespace.String
// 解密敏感数据
// 解密敏感数据（检测 kubeconfig 格式则跳过解密）
if encryptedCAData.Valid {
cluster.CAData, _ = r.encryptor.Decrypt(encryptedCAData.String)
cluster.CAData = r.decryptIfNeeded(encryptedCAData.String, "ca_data")
}
if encryptedCertData.Valid {
cluster.CertData, _ = r.encryptor.Decrypt(encryptedCertData.String)
cluster.CertData = r.decryptIfNeeded(encryptedCertData.String, "cert_data")
}
if encryptedKeyData.Valid {
cluster.KeyData, _ = r.encryptor.Decrypt(encryptedKeyData.String)
cluster.KeyData = r.decryptIfNeeded(encryptedKeyData.String, "key_data")
}
if encryptedToken.Valid {
cluster.Token, _ = r.encryptor.Decrypt(encryptedToken.String)
cluster.Token = r.decryptIfNeeded(encryptedToken.String, "token")
}
clusters = append(clusters, cluster)

127
backend/internal/adapter/output/persistence/postgres/storage_repository.go
View File

@ -12,6 +12,14 @@ import (
"github.com/ocdp/cluster-service/internal/domain/repository"
)
// sqlNullString converts empty string to sql.NullString for proper NULL handling
func sqlNullString(s string) interface{} {
if s == "" {
return sql.NullString{Valid: false}
}
return sql.NullString{String: s, Valid: true}
}
// StorageRepository PostgreSQL 存储后端仓储实现
type StorageRepository struct {
db *DB
@ -34,14 +42,15 @@ func (r *StorageRepository) Create(ctx context.Context, storage *entity.StorageB
}
query := `
INSERT INTO storage_backends (id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
INSERT INTO storage_backends (id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
`
_, err = r.db.conn.ExecContext(ctx, query,
storage.ID,
storage.WorkspaceID,
storage.OwnerID,
sqlNullString(storage.ClusterID),
sqlNullString(storage.OwnerID),
storage.Name,
storage.Type,
configJSON,
@ -62,17 +71,19 @@ func (r *StorageRepository) Create(ctx context.Context, storage *entity.StorageB
// GetByID 根据 ID 获取存储后端
func (r *StorageRepository) GetByID(ctx context.Context, id string) (*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE id = $1
`
storage := &entity.StorageBackend{}
var configJSON []byte
var wsID, clusterID, ownerID sql.NullString
err := r.db.conn.QueryRowContext(ctx, query, id).Scan(
&storage.ID,
&storage.WorkspaceID,
&storage.OwnerID,
&wsID,
&clusterID,
&ownerID,
&storage.Name,
&storage.Type,
&configJSON,
@ -82,6 +93,9 @@ func (r *StorageRepository) GetByID(ctx context.Context, id string) (*entity.Sto
&storage.CreatedAt,
&storage.UpdatedAt,
)
storage.WorkspaceID = wsID.String
storage.ClusterID = clusterID.String
storage.OwnerID = ownerID.String
if err == sql.ErrNoRows {
return nil, entity.ErrStorageNotFound
@ -100,17 +114,19 @@ func (r *StorageRepository) GetByID(ctx context.Context, id string) (*entity.Sto
// GetByName 根据名称获取存储后端
func (r *StorageRepository) GetByName(ctx context.Context, workspaceID, name string) (*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE workspace_id = $1 AND name = $2
`
storage := &entity.StorageBackend{}
var configJSON []byte
var wsID, clusterID, ownerID sql.NullString
err := r.db.conn.QueryRowContext(ctx, query, workspaceID, name).Scan(
&storage.ID,
&storage.WorkspaceID,
&storage.OwnerID,
&wsID,
&clusterID,
&ownerID,
&storage.Name,
&storage.Type,
&configJSON,
@ -120,6 +136,9 @@ func (r *StorageRepository) GetByName(ctx context.Context, workspaceID, name str
&storage.CreatedAt,
&storage.UpdatedAt,
)
storage.WorkspaceID = wsID.String
storage.ClusterID = clusterID.String
storage.OwnerID = ownerID.String
if err == sql.ErrNoRows {
return nil, entity.ErrStorageNotFound
@ -138,7 +157,7 @@ func (r *StorageRepository) GetByName(ctx context.Context, workspaceID, name str
// GetByWorkspace 获取 workspace 的所有存储后端
func (r *StorageRepository) GetByWorkspace(ctx context.Context, workspaceID string) ([]*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE workspace_id = $1 OR is_shared = TRUE
ORDER BY is_default DESC, name
@ -156,7 +175,7 @@ func (r *StorageRepository) GetByWorkspace(ctx context.Context, workspaceID stri
// GetShared 获取所有共享存储后端
func (r *StorageRepository) GetShared(ctx context.Context) ([]*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE is_shared = TRUE
ORDER BY name
@ -174,7 +193,7 @@ func (r *StorageRepository) GetShared(ctx context.Context) ([]*entity.StorageBac
// GetDefault 获取 workspace 的默认存储后端
func (r *StorageRepository) GetDefault(ctx context.Context, workspaceID string) (*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE workspace_id = $1 AND is_default = TRUE
LIMIT 1
@ -182,10 +201,12 @@ func (r *StorageRepository) GetDefault(ctx context.Context, workspaceID string)
storage := &entity.StorageBackend{}
var configJSON []byte
var wsID, clusterID, ownerID sql.NullString
err := r.db.conn.QueryRowContext(ctx, query, workspaceID).Scan(
&storage.ID,
&storage.WorkspaceID,
&storage.OwnerID,
&wsID,
&clusterID,
&ownerID,
&storage.Name,
&storage.Type,
&configJSON,
@ -195,6 +216,9 @@ func (r *StorageRepository) GetDefault(ctx context.Context, workspaceID string)
&storage.CreatedAt,
&storage.UpdatedAt,
)
storage.WorkspaceID = wsID.String
storage.ClusterID = clusterID.String
storage.OwnerID = ownerID.String
if err == sql.ErrNoRows {
return nil, nil
@ -210,6 +234,68 @@ func (r *StorageRepository) GetDefault(ctx context.Context, workspaceID string)
return storage, nil
}
// GetByCluster 获取 cluster 关联的存储后端列表
func (r *StorageRepository) GetByCluster(ctx context.Context, clusterID string) ([]*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE cluster_id = $1
ORDER BY is_default DESC, name
`
rows, err := r.db.conn.QueryContext(ctx, query, clusterID)
if err != nil {
return nil, fmt.Errorf("failed to list storage by cluster: %w", err)
}
defer rows.Close()
return r.scanStorages(rows)
}
// GetDefaultByCluster 获取 cluster 的默认存储后端
func (r *StorageRepository) GetDefaultByCluster(ctx context.Context, clusterID string) (*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
WHERE cluster_id = $1 AND is_default = TRUE
LIMIT 1
`
storage := &entity.StorageBackend{}
var configJSON []byte
var wsID, clusterIDNull, ownerID sql.NullString
err := r.db.conn.QueryRowContext(ctx, query, clusterID).Scan(
&storage.ID,
&wsID,
&clusterIDNull,
&ownerID,
&storage.Name,
&storage.Type,
&configJSON,
&storage.Description,
&storage.IsDefault,
&storage.IsShared,
&storage.CreatedAt,
&storage.UpdatedAt,
)
storage.WorkspaceID = wsID.String
storage.ClusterID = clusterIDNull.String
storage.OwnerID = ownerID.String
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("failed to get default storage by cluster: %w", err)
}
if err := json.Unmarshal(configJSON, &storage.Config); err != nil {
return nil, fmt.Errorf("failed to unmarshal config: %w", err)
}
return storage, nil
}
// Update 更新存储后端
func (r *StorageRepository) Update(ctx context.Context, storage *entity.StorageBackend) error {
storage.UpdatedAt = time.Now()
@ -221,8 +307,8 @@ func (r *StorageRepository) Update(ctx context.Context, storage *entity.StorageB
query := `
UPDATE storage_backends
SET name = $1, type = $2, config = $3, description = $4, is_default = $5, is_shared = $6, updated_at = $7
WHERE id = $8
SET name = $1, type = $2, config = $3, description = $4, is_default = $5, is_shared = $6, cluster_id = $7, updated_at = $8
WHERE id = $9
`
result, err := r.db.conn.ExecContext(ctx, query,
@ -232,6 +318,7 @@ func (r *StorageRepository) Update(ctx context.Context, storage *entity.StorageB
storage.Description,
storage.IsDefault,
storage.IsShared,
sqlNullString(storage.ClusterID),
storage.UpdatedAt,
storage.ID,
)
@ -276,7 +363,7 @@ func (r *StorageRepository) Delete(ctx context.Context, id string) error {
// List 列出所有存储后端（管理员用）
func (r *StorageRepository) List(ctx context.Context) ([]*entity.StorageBackend, error) {
query := `
SELECT id, workspace_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
SELECT id, workspace_id, cluster_id, owner_id, name, type, config, description, is_default, is_shared, created_at, updated_at
FROM storage_backends
ORDER BY workspace_id, name
`
@ -296,9 +383,11 @@ func (r *StorageRepository) scanStorages(rows *sql.Rows) ([]*entity.StorageBacke
for rows.Next() {
storage := &entity.StorageBackend{}
var configJSON []byte
var wsID, clusterID sql.NullString
err := rows.Scan(
&storage.ID,
&storage.WorkspaceID,
&wsID,
&clusterID,
&storage.OwnerID,
&storage.Name,
&storage.Type,
@ -312,6 +401,8 @@ func (r *StorageRepository) scanStorages(rows *sql.Rows) ([]*entity.StorageBacke
if err != nil {
return nil, fmt.Errorf("failed to scan storage: %w", err)
}
storage.WorkspaceID = wsID.String
storage.ClusterID = clusterID.String
if err := json.Unmarshal(configJSON, &storage.Config); err != nil {
return nil, fmt.Errorf("failed to unmarshal config: %w", err)
}