diff --git a/ansible/mysql-deploy-callback.yml b/ansible/mysql-deploy-callback.yml index 434a113..befff5f 100644 --- a/ansible/mysql-deploy-callback.yml +++ b/ansible/mysql-deploy-callback.yml @@ -63,9 +63,9 @@ # --- expected node count per topology --- mysql_expected_hosts: "{{ {'standalone': 1, 'primary_replica': 2, 'mgr_3': 3}[topology | default('standalone')] }}" - # --- secrets (injected via environment) --- - mysql_root_password_value: "{{ lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD') }}" - mysql_admin_password_value: "{{ lookup('ansible.builtin.env', 'XINFRA_MYSQL_ADMIN_PASSWORD') }}" + # --- secrets (prefer launch extra_vars; fall back to AWX credential-injected env) --- + mysql_root_password_value: "{{ mysql_root_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD'), true) }}" + mysql_admin_password_value: "{{ mysql_admin_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ADMIN_PASSWORD'), true) }}" # --- platform callback --- delivery_callback_url_value: "{{ delivery_callback_url | default('') }}" diff --git a/ansible/mysql-deploy.yml b/ansible/mysql-deploy.yml index c66e2c5..28e5ec9 100644 --- a/ansible/mysql-deploy.yml +++ b/ansible/mysql-deploy.yml @@ -63,9 +63,9 @@ # --- expected node count per topology --- mysql_expected_hosts: "{{ {'standalone': 1, 'primary_replica': 2, 'mgr_3': 3}[topology | default('standalone')] }}" - # --- secrets (injected via environment) --- - mysql_root_password_value: "{{ lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD') }}" - mysql_admin_password_value: "{{ lookup('ansible.builtin.env', 'XINFRA_MYSQL_ADMIN_PASSWORD') }}" + # --- secrets (prefer launch extra_vars; fall back to AWX credential-injected env) --- + mysql_root_password_value: "{{ mysql_root_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ROOT_PASSWORD'), true) }}" + mysql_admin_password_value: "{{ mysql_admin_password | default(lookup('ansible.builtin.env', 'XINFRA_MYSQL_ADMIN_PASSWORD'), true) }}" pre_tasks: - name: Validate delivery parameters diff --git a/server/.env.example b/server/.env.example index e29aab7..ec05d4e 100644 --- a/server/.env.example +++ b/server/.env.example @@ -6,19 +6,42 @@ OIDC_AUTHORIZATION_ENDPOINT=http://localhost:8080/auth/oauth/authorize OIDC_TOKEN_ENDPOINT=http://localhost:8080/auth/oauth/token OIDC_USERINFO_ENDPOINT=http://localhost:8080/auth/oauth/userinfo OIDC_JWKS_URI=http://localhost:8080/auth/oauth/jwks +OIDC_CLOUDDM_CLIENT_ID=clouddm +OIDC_CLOUDDM_CLIENT_SECRET=change-this-clouddm-client-secret +OIDC_CLOUDDM_REDIRECT_URI= CLOUDDM_TARGET_URL=http://authserver-nginx/internal/clouddm +WAYEN_LOGIN_URL= +WAYEN_TARGET_URL= +WAYEN_USERNAME_KEY=username +WAYEN_PASSWORD_KEY=password +WAYEN_LOGIN_FORMAT=query +WAYEN_LOGIN_VALUE=username +WAYEN_OAUTH_REF=/portal/namespace/1/app +WAYEN_OAUTH_LOGIN_URL= WAYNE_API_BASE_URL=http://wayne-backend:8080 WAYNE_ADMIN_USERNAME=admin WAYNE_ADMIN_PASSWORD=change-this-wayne-admin-password WAYNE_TOKEN_TTL_MINUTES=1440 +WAYNE_INTERNAL_API_BASE_URL= +WAYNE_SERVICE_NAME=xinfra +WAYNE_SERVICE_API_SECRET_KEY= +OAUTH_WAYNE_CLIENT_ID=wayne +OAUTH_WAYNE_CLIENT_SECRET=change-this-wayne-client-secret +OAUTH_WAYNE_REDIRECT_URI= +OAUTH_CODE_TTL_SECONDS=120 MYSQL_DSN=auth:auth@tcp(127.0.0.1:3000)/authserver?charset=utf8mb4&parseTime=True&loc=Local AUTO_MIGRATE=true +SSO_ENABLED=true + +BOOTSTRAP_ADMIN_USERNAME=admin +BOOTSTRAP_ADMIN_PASSWORD=change-this-bootstrap-admin-password # MySQL service delivery (AWX is required when the scheduler is enabled) DELIVERY_SCHEDULER_ENABLED=false DELIVERY_DISPATCH_SECONDS=5 DELIVERY_CALLBACK_BASE_URL=http://authserver-backend.authserver.svc.cluster.local:8083 +DELIVERY_CREDENTIAL_SECRET= DELIVERY_RESERVATION_TTL_MINUTES=120 DELIVERY_GLOBAL_LIMIT=2 DELIVERY_TARGET_LIMIT=2 @@ -47,5 +70,11 @@ SAML_ENTITY_ID=http://localhost:8080/api/v1/saml/metadata SAML_ACS_URL=http://localhost:8080/api/v1/saml/acs SAML_IDP_METADATA_URL=http://sso-internal.dev.qiniu.io/saml2/meta SAML_LOGOUT_URL=http://sso-internal.dev.qiniu.io/signout +SAML_IDP_METADATA_CACHE_FILE= SAML_SP_CERT_FILE=certs/sp.crt SAML_SP_KEY_FILE=certs/sp.key + +# SINA CMDB machine resources +SINA_BASE_URL=https://sinai.qiniu.io:443 +SINA_USERNAME= +SINA_PASSWORD= diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 8254286..f5733ce 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -71,11 +71,15 @@ type Config struct { DeliverySchedulerEnabled bool DeliveryDispatchSeconds int DeliveryCallbackBaseURL string + DeliveryCredentialSecret string ReservationTTLMinutes int DeliveryGlobalLimit int DeliveryTargetLimit int DeliveryHostInstanceLimit int DeliveryDataDisks []string + SINABaseURL string + SINAUsername string + SINAPassword string } func Load() Config { @@ -145,11 +149,15 @@ func Load() Config { DeliverySchedulerEnabled: envBool("DELIVERY_SCHEDULER_ENABLED", false), DeliveryDispatchSeconds: envInt("DELIVERY_DISPATCH_SECONDS", 5), DeliveryCallbackBaseURL: trimURL(env("DELIVERY_CALLBACK_BASE_URL", publicBaseURL)), + DeliveryCredentialSecret: env("DELIVERY_CREDENTIAL_SECRET", env("JWT_SECRET", "change-this-secret")), ReservationTTLMinutes: envInt("DELIVERY_RESERVATION_TTL_MINUTES", 120), DeliveryGlobalLimit: envInt("DELIVERY_GLOBAL_LIMIT", 2), DeliveryTargetLimit: envInt("DELIVERY_TARGET_LIMIT", 2), DeliveryHostInstanceLimit: envInt("DELIVERY_HOST_INSTANCE_LIMIT", 4), DeliveryDataDisks: splitCSV(env("DELIVERY_DATA_DISKS", "/data,/disk1,/mnt,/opt/mysql-delivery")), + SINABaseURL: trimURL(env("SINA_BASE_URL", "https://sinai.qiniu.io:443")), + SINAUsername: env("SINA_USERNAME", ""), + SINAPassword: env("SINA_PASSWORD", ""), } } diff --git a/server/internal/database/database.go b/server/internal/database/database.go index 3123ba0..d2287de 100644 --- a/server/internal/database/database.go +++ b/server/internal/database/database.go @@ -24,10 +24,13 @@ func AutoMigrate(db *gorm.DB) error { &model.ResourceQuota{}, &model.DeliveryTask{}, &model.ResourceReservation{}, - &model.MySQLInstance{}, + &model.DeploymentResult{}, + &model.DeploymentCredential{}, &model.ResourceUsage{}, &model.ExecutionJob{}, &model.RollbackJob{}, &model.TaskEvent{}, + &model.MachineResource{}, + &model.MachineSyncState{}, ) } diff --git a/server/internal/handler/container_service.go b/server/internal/handler/container_service.go index fb8e193..ccfaa28 100644 --- a/server/internal/handler/container_service.go +++ b/server/internal/handler/container_service.go @@ -37,6 +37,14 @@ func (h *ContainerServiceHandler) Workloads(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"items": data.Workloads, "summary": data.Summary}) } +func (h *ContainerServiceHandler) Clusters(c *gin.Context) { + data, ok := h.loadData(c) + if !ok { + return + } + c.JSON(http.StatusOK, gin.H{"items": data.Clusters, "nodes": data.Nodes, "summary": data.Summary}) +} + func (h *ContainerServiceHandler) loadData(c *gin.Context) (*service.ContainerServiceData, bool) { claims, ok := CurrentClaims(c) if !ok { diff --git a/server/internal/handler/delivery.go b/server/internal/handler/delivery.go index e6fe1c1..09387fe 100644 --- a/server/internal/handler/delivery.go +++ b/server/internal/handler/delivery.go @@ -140,6 +140,47 @@ func (h *DeliveryHandler) List(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"items": items}) } +func (h *DeliveryHandler) MySQLServiceLedger(c *gin.Context) { + claims, ok := CurrentClaims(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "missing current user"}) + return + } + businessLineID, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil || businessLineID == 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid business line id"}) + return + } + items, err := h.service.ListMySQLServiceLedger(c.Request.Context(), claims.UserID, claims.IsAdmin, businessLineID) + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "business line not found"}) + return + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"items": items}) +} + +func (h *DeliveryHandler) RevealCredentials(c *gin.Context) { + claims, ok := CurrentClaims(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "missing current user"}) + return + } + items, err := h.service.RevealDeploymentCredentials(c.Request.Context(), c.Param("id"), claims.UserID, claims.IsAdmin) + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "credentials not found or already viewed"}) + return + } + if err != nil { + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"items": items}) +} + // Get 获取交付任务详情 // @Summary 获取交付任务详情 // @Description 返回指定任务的详细信息及事件流 diff --git a/server/internal/handler/machine.go b/server/internal/handler/machine.go new file mode 100644 index 0000000..0d21c03 --- /dev/null +++ b/server/internal/handler/machine.go @@ -0,0 +1,69 @@ +package handler + +import ( + "net/http" + "strconv" + + "github.com/1024XEngineer/xinfra/server/internal/service" + + "github.com/gin-gonic/gin" +) + +type MachineHandler struct { + machines *service.MachineService +} + +func NewMachineHandler(machines *service.MachineService) *MachineHandler { + return &MachineHandler{machines: machines} +} + +func (h *MachineHandler) Overview(c *gin.Context) { + overview, err := h.machines.Overview(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, overview) +} + +func (h *MachineHandler) List(c *gin.Context) { + resources, err := h.machines.List(c.Request.Context(), service.MachineListQuery{ + Page: queryInt(c, "page", 1), + Size: queryInt(c, "size", 20), + Hostname: c.Query("hostname"), + AssetNumber: c.Query("assetNumber"), + Type: c.Query("type"), + Location: c.Query("location"), + IP: c.Query("ip"), + Spec: c.Query("spec"), + BusinessLine: c.Query("businessLine"), + Source: c.Query("source"), + Status: c.Query("status"), + }) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, resources) +} + +func (h *MachineHandler) Sync(c *gin.Context) { + state, err := h.machines.SyncNow(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error(), "state": state}) + return + } + c.JSON(http.StatusOK, state) +} + +func queryInt(c *gin.Context, key string, fallback int) int { + value := c.Query(key) + if value == "" { + return fallback + } + n, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return n +} diff --git a/server/internal/model/delivery.go b/server/internal/model/delivery.go index b5ec692..86dc539 100644 --- a/server/internal/model/delivery.go +++ b/server/internal/model/delivery.go @@ -87,23 +87,44 @@ type ResourceUsage struct { ReleasedAt *time.Time `json:"released_at,omitempty"` } -type MySQLInstance struct { +type DeploymentResult struct { ID uint64 `gorm:"primaryKey" json:"id"` TaskID string `gorm:"size:36;not null;uniqueIndex" json:"task_id"` BusinessLineID uint64 `gorm:"not null;index" json:"business_line_id"` - TargetID uint64 `gorm:"not null;index;uniqueIndex:idx_mysql_target_name,priority:1" json:"target_id"` - Namespace string `gorm:"size:63;not null" json:"namespace"` - Name string `gorm:"size:63;not null;uniqueIndex:idx_mysql_target_name,priority:2" json:"name"` - NodeName string `gorm:"size:128" json:"node_name"` - Host string `gorm:"size:255;not null" json:"host"` - Port int `gorm:"not null;default:3306" json:"port"` - Version string `gorm:"size:32;not null" json:"version"` + Component string `gorm:"size:32;not null;index" json:"component"` + ServiceType string `gorm:"size:32;not null;index" json:"service_type"` + InstanceName string `gorm:"size:128;not null;index" json:"instance_name"` + Namespace string `gorm:"size:128;not null;default:'';index" json:"namespace"` + TargetID uint64 `gorm:"not null;index" json:"target_id"` + NodeName string `gorm:"size:128;not null;default:''" json:"node_name"` + Host string `gorm:"size:255;not null;default:''" json:"host"` + Port int `gorm:"not null;default:0" json:"port"` + Version string `gorm:"size:64;not null;default:''" json:"version"` Status string `gorm:"size:32;not null;index" json:"status"` - CloudDMID string `gorm:"size:128" json:"clouddm_id,omitempty"` + Metadata string `gorm:"type:json" json:"metadata"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } +type DeploymentCredential struct { + ID uint64 `gorm:"primaryKey" json:"id"` + TaskID string `gorm:"size:36;not null;index" json:"task_id"` + DeploymentResultID uint64 `gorm:"not null;default:0;index" json:"deployment_result_id"` + BusinessLineID uint64 `gorm:"not null;index" json:"business_line_id"` + Component string `gorm:"size:32;not null;index" json:"component"` + InstanceName string `gorm:"size:128;not null;index" json:"instance_name"` + Username string `gorm:"size:128;not null" json:"username"` + AccountHost string `gorm:"size:128;not null;default:''" json:"account_host"` + Ciphertext string `gorm:"type:text;not null" json:"-"` + Nonce string `gorm:"size:64;not null" json:"-"` + Status string `gorm:"size:32;not null;index" json:"status"` + Source string `gorm:"size:64;not null;default:'user_input'" json:"source"` + ViewedBy uint64 `gorm:"not null;default:0" json:"viewed_by"` + ViewedAt *time.Time `json:"viewed_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + type ExecutionJob struct { ID uint64 `gorm:"primaryKey" json:"id"` TaskID string `gorm:"size:36;not null;uniqueIndex" json:"task_id"` diff --git a/server/internal/model/machine.go b/server/internal/model/machine.go new file mode 100644 index 0000000..834795b --- /dev/null +++ b/server/internal/model/machine.go @@ -0,0 +1,54 @@ +package model + +import "time" + +type MachineResource struct { + ID uint64 `gorm:"primaryKey" json:"id"` + SinaID int64 `gorm:"not null;uniqueIndex" json:"sina_id"` + CIID string `gorm:"size:64;not null;default:'';index" json:"ci_id"` + Hostname string `gorm:"size:255;not null;default:'';index" json:"hostname"` + Name string `gorm:"size:255;not null;default:''" json:"name"` + AssetNumber string `gorm:"size:128;not null;default:'';index" json:"asset_number"` + IDCNumber string `gorm:"size:128;not null;default:''" json:"idc_number"` + ResourceType string `gorm:"size:64;not null;default:'';index" json:"resource_type"` + LocationName string `gorm:"size:255;not null;default:'';index" json:"location_name"` + LocationAlias string `gorm:"size:255;not null;default:''" json:"location_alias"` + IntranetIP string `gorm:"size:128;not null;default:'';index" json:"intranet_ip"` + Spec string `gorm:"size:255;not null;default:''" json:"spec"` + BusinessLine string `gorm:"size:128;not null;default:'';index" json:"business_line"` + Source string `gorm:"size:64;not null;default:'';index" json:"source"` + Status string `gorm:"size:64;not null;default:'';index" json:"status"` + Brand string `gorm:"size:128;not null;default:''" json:"brand"` + Model string `gorm:"size:128;not null;default:''" json:"model"` + SerialNumber string `gorm:"size:128;not null;default:'';index" json:"serial_number"` + PowerStatus string `gorm:"size:64;not null;default:''" json:"power_status"` + CPUCores int64 `gorm:"not null;default:0" json:"cpu_cores"` + CPUName string `gorm:"size:255;not null;default:''" json:"cpu_name"` + Memory string `gorm:"size:255;not null;default:''" json:"memory"` + HardCapacity string `gorm:"size:255;not null;default:''" json:"hard_capacity"` + SSDCapacity string `gorm:"size:255;not null;default:''" json:"ssd_capacity"` + OOBIP string `gorm:"size:128;not null;default:''" json:"oob_ip"` + OSFamily string `gorm:"size:128;not null;default:''" json:"os_family"` + OSVersion string `gorm:"size:128;not null;default:''" json:"os_version"` + CollectTime *time.Time `gorm:"index" json:"collect_time"` + SinaUpdatedAt *time.Time `gorm:"index" json:"sina_updated_at"` + RawJSON string `gorm:"type:longtext" json:"-"` + LastSyncedAt time.Time `gorm:"not null;index" json:"last_synced_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +type MachineSyncState struct { + ID uint64 `gorm:"primaryKey" json:"id"` + Source string `gorm:"size:64;not null;uniqueIndex" json:"source"` + Status string `gorm:"size:32;not null;default:'idle';index" json:"status"` + LastStartedAt *time.Time `json:"last_started_at"` + LastFinishedAt *time.Time `json:"last_finished_at"` + LastError string `gorm:"type:text" json:"last_error"` + Total int64 `gorm:"not null;default:0" json:"total"` + Fetched int64 `gorm:"not null;default:0" json:"fetched"` + CreatedCount int64 `gorm:"not null;default:0" json:"created_count"` + UpdatedCount int64 `gorm:"not null;default:0" json:"updated_count"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} diff --git a/server/internal/router/router.go b/server/internal/router/router.go index 313daf4..ae0b365 100644 --- a/server/internal/router/router.go +++ b/server/internal/router/router.go @@ -72,9 +72,11 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { wayenService := service.NewWayenService(deps.Config, deps.DB) wayneRoleBindingService := service.NewWayneRoleBindingService(deps.Config, deps.DB) deliveryService := service.NewDeliveryService(deps.Config, deps.DB, auditService) + machineService := service.NewMachineService(deps.Config, deps.DB) if deps.Config.DeliverySchedulerEnabled { go deliveryService.Run(context.Background()) } + machineService.Run(context.Background()) healthHandler := handler.NewHealthHandler(deps.DB) authHandler := handler.NewAuthHandler(deps.Config, authService) @@ -90,6 +92,7 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { deliveryCallbackHandler := handler.NewDeliveryCallbackHandler(deliveryService, deps.Config.AWXWebhookToken) containerServiceHandler := handler.NewContainerServiceHandler(deps.DB, wayneRoleBindingService) taskLogHandler := handler.NewTaskLogHandler(deps.DB, deliveryService, wayneRoleBindingService) + machineHandler := handler.NewMachineHandler(machineService) r.GET("/healthz", healthHandler.Healthz) r.GET("/readyz", healthHandler.Readyz) @@ -143,19 +146,25 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) { protected.POST("/subsystem-auth/wayne/business-lines/:id/users/:userid/init", subsystemAuthHandler.InitWayneBusinessLineUser) protected.GET("/container-services/business-lines/:id/summary", containerServiceHandler.Summary) protected.GET("/container-services/business-lines/:id/workloads", containerServiceHandler.Workloads) + protected.GET("/container-services/business-lines/:id/clusters", containerServiceHandler.Clusters) protected.GET("/clouddm/login", clouddmHandler.Login) protected.GET("/delivery/targets", deliveryHandler.Targets) protected.GET("/delivery/targets/:target_id/hosts/:host/mount-paths", deliveryHandler.TargetHostMountPaths) protected.PUT("/delivery/quotas", deliveryHandler.UpsertQuota) protected.POST("/delivery/mysql", deliveryHandler.CreateMySQL) protected.GET("/delivery/tasks", deliveryHandler.List) + protected.GET("/delivery/business-lines/:id/mysql-services", deliveryHandler.MySQLServiceLedger) protected.GET("/delivery/tasks/:id", deliveryHandler.Get) protected.GET("/delivery/tasks/:id/stream", deliveryHandler.Stream) + protected.POST("/delivery/tasks/:id/credentials/reveal", deliveryHandler.RevealCredentials) protected.POST("/delivery/tasks/:id/cancel", deliveryHandler.Cancel) protected.POST("/delivery/tasks/:id/rollback/retry", deliveryHandler.RetryRollback) protected.POST("/delivery/tasks/:id/rollback/release", deliveryHandler.AcknowledgeRollbackRelease) protected.POST("/delivery/tasks/:id/clouddm/retry", deliveryHandler.RetryCloudDMRegistration) protected.GET("/task-logs", taskLogHandler.List) protected.GET("/task-logs/:id", taskLogHandler.Get) + protected.GET("/machines/overview", machineHandler.Overview) + protected.GET("/machines/resources", machineHandler.List) + protected.POST("/machines/sync", machineHandler.Sync) } } diff --git a/server/internal/service/container_service.go b/server/internal/service/container_service.go index 7218931..e773657 100644 --- a/server/internal/service/container_service.go +++ b/server/internal/service/container_service.go @@ -47,9 +47,38 @@ type ContainerWorkload struct { StatusClass string `json:"statusClass"` } +type ContainerCluster struct { + Name string `json:"name"` + Zone string `json:"zone"` + ZoneClass string `json:"zoneClass"` + Status string `json:"status"` + StatusClass string `json:"statusClass"` + Nodes int `json:"nodes"` + ReadyNodes int `json:"readyNodes"` + CPU int `json:"cpu"` + CPUClass string `json:"cpuClass"` + Version string `json:"version"` + Calico string `json:"calico"` +} + +type ContainerNode struct { + Cluster string `json:"cluster"` + Name string `json:"name"` + IP string `json:"ip"` + Label string `json:"label"` + Taint string `json:"taint"` + Spec string `json:"spec"` + CPU int `json:"cpu"` + CPUClass string `json:"cpuClass"` + Status string `json:"status"` + StatusClass string `json:"statusClass"` +} + type ContainerServiceData struct { Summary ContainerServiceSummary `json:"summary"` Workloads []ContainerWorkload `json:"items"` + Clusters []ContainerCluster `json:"clusters"` + Nodes []ContainerNode `json:"nodes"` } type ContainerBusinessLineQuery struct { @@ -81,7 +110,7 @@ func (s *WayneRoleBindingService) ContainerServiceData(ctx context.Context, quer Namespaces: len(query.Namespaces), } workloads := make([]ContainerWorkload, 0) - clusterNodes := map[string]wayneNodeSummary{} + clusterNodes := map[string]wayneNodeData{} seenWorkloads := map[string]struct{}{} errors := make([]string, 0) @@ -142,7 +171,9 @@ func (s *WayneRoleBindingService) ContainerServiceData(ctx context.Context, quer } } - for _, nodes := range clusterNodes { + clusters := make([]ContainerCluster, 0, len(clusterNodes)) + allNodes := make([]ContainerNode, 0) + for clusterName, nodes := range clusterNodes { summary.Nodes += nodes.Total summary.ReadyNodes += nodes.Ready summary.SchedulableNodes += nodes.Schedulable @@ -150,6 +181,10 @@ func (s *WayneRoleBindingService) ContainerServiceData(ctx context.Context, quer summary.CPUTotal += nodes.CPUTotal summary.MemoryUsed += nodes.MemoryUsed summary.MemoryTotal += nodes.MemoryTotal + clusters = append(clusters, containerClusterFromWayne(clusterName, nodes)) + for _, node := range nodes.Nodes { + allNodes = append(allNodes, containerNodeFromWayne(clusterName, node, query.BusinessLineName)) + } } summary.Clusters = len(clusterNodes) summary.Workloads = len(workloads) @@ -157,7 +192,9 @@ func (s *WayneRoleBindingService) ContainerServiceData(ctx context.Context, quer summary.MemoryPercent = ratioPercent(summary.MemoryUsed, summary.MemoryTotal) summary.Errors = errors - return &ContainerServiceData{Summary: summary, Workloads: workloads}, nil + sortContainerClusters(clusters) + sortContainerNodes(allNodes) + return &ContainerServiceData{Summary: summary, Workloads: workloads, Clusters: clusters, Nodes: allNodes}, nil } func (s *WayneRoleBindingService) ListDeploymentHistories(ctx context.Context, namespaces []model.BusinessLineWayneNamespace, limit int) ([]WayneDeploymentHistory, error) { @@ -264,10 +301,10 @@ func (s *WayneRoleBindingService) listWaynePublishHistories(ctx context.Context, return parseWaynePublishHistories(result.Body) } -func (s *WayneRoleBindingService) getWayneNodes(ctx context.Context, cluster string) (wayneNodeSummary, error) { +func (s *WayneRoleBindingService) getWayneNodes(ctx context.Context, cluster string) (wayneNodeData, error) { result, err := s.callRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/kubernetes/nodes/clusters/%s", url.PathEscape(cluster)), nil) if err != nil { - return wayneNodeSummary{}, err + return wayneNodeData{}, err } return parseWayneNodeSummary(result.Body) } @@ -323,7 +360,7 @@ type wayneDeployment struct { Name string `json:"name"` } -type wayneNodeSummary struct { +type wayneNodeData struct { Total int Ready int Schedulable int @@ -331,6 +368,23 @@ type wayneNodeSummary struct { CPUTotal float64 MemoryUsed float64 MemoryTotal float64 + Nodes []wayneNode +} + +type wayneNode struct { + Name string `json:"name"` + Labels map[string]string `json:"labels"` + Spec struct { + Unschedulable bool `json:"unschedulable"` + Taints []map[string]any `json:"taints"` + Ready string `json:"ready"` + } `json:"spec"` + Status struct { + Capacity map[string]string `json:"capacity"` + NodeInfo struct { + KubeletVersion string `json:"kubeletVersion"` + } `json:"nodeInfo"` + } `json:"status"` } func parseWayneNamespaceDetail(body []byte) (wayneNamespaceDetail, error) { @@ -420,7 +474,7 @@ func sortWayneDeploymentHistories(items []WayneDeploymentHistory) { } } -func parseWayneNodeSummary(body []byte) (wayneNodeSummary, error) { +func parseWayneNodeSummary(body []byte) (wayneNodeData, error) { var wrapped struct { Data struct { NodeSummary struct { @@ -436,12 +490,13 @@ func parseWayneNodeSummary(body []byte) (wayneNodeSummary, error) { Total float64 `json:"total"` Used float64 `json:"used"` } `json:"memorySummary"` + Nodes []wayneNode `json:"nodes"` } `json:"data"` } if err := json.Unmarshal(body, &wrapped); err != nil { - return wayneNodeSummary{}, err + return wayneNodeData{}, err } - return wayneNodeSummary{ + return wayneNodeData{ Total: wrapped.Data.NodeSummary.Total, Ready: wrapped.Data.NodeSummary.Ready, Schedulable: wrapped.Data.NodeSummary.Schedulable, @@ -449,9 +504,159 @@ func parseWayneNodeSummary(body []byte) (wayneNodeSummary, error) { CPUTotal: wrapped.Data.CPUSummary.Total, MemoryUsed: wrapped.Data.MemorySummary.Used, MemoryTotal: wrapped.Data.MemorySummary.Total, + Nodes: wrapped.Data.Nodes, }, nil } +func containerClusterFromWayne(name string, data wayneNodeData) ContainerCluster { + status := "健康" + statusClass := "ok" + if data.Total == 0 { + status = "无节点" + statusClass = "idle" + } else if data.Ready < data.Total { + status = strconv.Itoa(data.Total-data.Ready) + " 节点告警" + statusClass = "warn" + } + cpu := ratioPercent(data.CPUUsed, data.CPUTotal) + version := "-" + for _, node := range data.Nodes { + if strings.TrimSpace(node.Status.NodeInfo.KubeletVersion) != "" { + version = strings.TrimSpace(node.Status.NodeInfo.KubeletVersion) + break + } + } + return ContainerCluster{ + Name: name, + Zone: "-", + Status: status, + StatusClass: statusClass, + Nodes: data.Total, + ReadyNodes: data.Ready, + CPU: cpu, + CPUClass: warnClass(cpu), + Version: version, + Calico: "-", + } +} + +func containerNodeFromWayne(cluster string, node wayneNode, businessLineName string) ContainerNode { + cpu := 0 + if node.Status.Capacity != nil { + cpu, _ = strconv.Atoi(strings.TrimSpace(node.Status.Capacity["cpu"])) + } + memory := strings.TrimSpace(node.Status.Capacity["memory"]) + spec := "-" + if cpu > 0 && memory != "" { + spec = fmt.Sprintf("%dC/%sG", cpu, memory) + } else if cpu > 0 { + spec = fmt.Sprintf("%dC", cpu) + } else if memory != "" { + spec = memory + "G" + } + ready := strings.EqualFold(strings.TrimSpace(node.Spec.Ready), "true") + status := "Ready" + statusClass := "ok" + if !ready { + status = "NotReady" + statusClass = "warn" + } else if node.Spec.Unschedulable { + status = "Ready · 不可调度" + statusClass = "warn" + } + label := businessLineLabel(node.Labels, businessLineName) + return ContainerNode{ + Cluster: cluster, + Name: node.Name, + IP: firstNodeIP(node.Labels), + Label: label, + Taint: formatNodeTaints(node.Spec.Taints), + Spec: spec, + CPU: 0, + CPUClass: "", + Status: status, + StatusClass: statusClass, + } +} + +func warnClass(value int) string { + if value >= 80 { + return "warn" + } + return "" +} + +func businessLineLabel(labels map[string]string, businessLineName string) string { + for _, key := range []string{"business-line", "business_line", "xinfra/business-line"} { + if value := strings.TrimSpace(labels[key]); value != "" { + return key + "=" + value + } + } + if businessLineName != "" { + return "business-line=" + businessLineName + } + return "-" +} + +func firstNodeIP(labels map[string]string) string { + for _, key := range []string{"kubernetes.io/hostname", "internal-ip", "node-ip"} { + if value := strings.TrimSpace(labels[key]); value != "" && strings.Contains(value, ".") { + return value + } + } + return "-" +} + +func formatNodeTaints(taints []map[string]any) string { + if len(taints) == 0 { + return "-" + } + parts := make([]string, 0, len(taints)) + for _, taint := range taints { + key := strings.TrimSpace(stringFromAny(taint["key"])) + value := strings.TrimSpace(stringFromAny(taint["value"])) + effect := strings.TrimSpace(stringFromAny(taint["effect"])) + text := key + if value != "" { + text += "=" + value + } + if effect != "" { + text += ":" + effect + } + if text != "" { + parts = append(parts, text) + } + } + if len(parts) == 0 { + return "-" + } + return strings.Join(parts, ", ") +} + +func sortContainerClusters(items []ContainerCluster) { + for i := 1; i < len(items); i++ { + item := items[i] + j := i - 1 + for j >= 0 && items[j].Name > item.Name { + items[j+1] = items[j] + j-- + } + items[j+1] = item + } +} + +func sortContainerNodes(items []ContainerNode) { + for i := 1; i < len(items); i++ { + item := items[i] + j := i - 1 + for j >= 0 && (items[j].Cluster > item.Cluster || (items[j].Cluster == item.Cluster && items[j].Name > item.Name)) { + items[j+1] = items[j] + j-- + } + items[j+1] = item + } +} + func containerWorkloadFromMap(item map[string]any, cluster string, namespace string, businessLineName string) ContainerWorkload { name := firstString(item, "name", "objectMeta.name", "metadata.name") ready := firstInt(item, "pods.current", "readyReplicas", "status.readyReplicas", "availableReplicas", "status.availableReplicas") diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 83efd21..50f34af 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -3,8 +3,11 @@ package service import ( "bytes" "context" + "crypto/aes" + "crypto/cipher" "crypto/rand" "crypto/sha256" + "encoding/base64" "encoding/hex" "encoding/json" "errors" @@ -60,6 +63,8 @@ type MySQLDeliveryInput struct { LongQueryTime float64 `json:"long_query_time"` BinlogExpireLogsSeconds int64 `json:"binlog_expire_logs_seconds"` MaxBinlogSize string `json:"max_binlog_size"` + MySQLRootPassword string `json:"mysql_root_password"` + MySQLAdminPassword string `json:"mysql_admin_password"` } type deliveryPayload struct { @@ -110,6 +115,25 @@ type DeliveryTaskSnapshot struct { Events []model.TaskEvent `json:"events"` } +type MySQLServiceLedgerItem struct { + Name string `json:"name"` + Datacenter string `json:"datacenter"` + BusinessTag string `json:"business_tag"` + Instances int `json:"instances"` + Healthy int `json:"healthy"` + Address string `json:"address"` + Status string `json:"status"` + StatusClass string `json:"status_class"` + Version string `json:"version"` + Namespace string `json:"namespace"` +} + +type DeploymentCredentialView struct { + Username string `json:"username"` + Host string `json:"host"` + Password string `json:"password"` +} + // targetMetadata describes the native VM候选节点池以及部署形态，由 AWX inventory hosts 动态组装。 type targetMetadata struct { Topology string `json:"topology"` @@ -211,6 +235,52 @@ func bytesToGi(bytes int64) int64 { return bytes / 1073741824 } +func credentialKey(secret string) []byte { + sum := sha256.Sum256([]byte(strings.TrimSpace(secret))) + return sum[:] +} + +func encryptCredential(secret, plain string) (string, string, error) { + block, err := aes.NewCipher(credentialKey(secret)) + if err != nil { + return "", "", err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return "", "", err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := rand.Read(nonce); err != nil { + return "", "", err + } + ciphertext := gcm.Seal(nil, nonce, []byte(plain), nil) + return base64.StdEncoding.EncodeToString(ciphertext), base64.StdEncoding.EncodeToString(nonce), nil +} + +func decryptCredential(secret, ciphertextValue, nonceValue string) (string, error) { + ciphertext, err := base64.StdEncoding.DecodeString(ciphertextValue) + if err != nil { + return "", err + } + nonce, err := base64.StdEncoding.DecodeString(nonceValue) + if err != nil { + return "", err + } + block, err := aes.NewCipher(credentialKey(secret)) + if err != nil { + return "", err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return "", err + } + plain, err := gcm.Open(nil, nonce, ciphertext, nil) + if err != nil { + return "", err + } + return string(plain), nil +} + // firstFreeHost 返回候选池中非失败任务数未达单机实例上限的第一个节点； // limit < 1 时按 1 兜底（退化为旧的一机一实例语义）。顺序遍历天然形成"先摊开、摊满一轮再叠加"。 func firstFreeHost(hosts []targetHost, occupied []string, limit int) *targetHost { @@ -406,6 +476,21 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin if err := validateDeliveryInput(input, s.cfg.DeliveryDataDisks); err != nil { return nil, false, err } + credentialInput := map[string]string{ + "root@localhost": strings.TrimSpace(input.MySQLRootPassword), + "xinfra_admin@%": strings.TrimSpace(input.MySQLAdminPassword), + } + hasCredentialInput := credentialInput["root@localhost"] != "" || credentialInput["xinfra_admin@%"] != "" + if hasCredentialInput { + if credentialInput["root@localhost"] == "" || credentialInput["xinfra_admin@%"] == "" { + return nil, false, fmt.Errorf("mysql_root_password and mysql_admin_password must be provided together") + } + if len(credentialInput["root@localhost"]) < 16 || len(credentialInput["xinfra_admin@%"]) < 16 { + return nil, false, fmt.Errorf("mysql passwords must be at least 16 characters") + } + } + input.MySQLRootPassword = "" + input.MySQLAdminPassword = "" var existing model.DeliveryTask if err := s.db.WithContext(ctx).Where("idempotency_key = ?", idempotencyKey).First(&existing).Error; err == nil { @@ -457,6 +542,7 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin ID: randomUUID(), BusinessLineID: input.BusinessLineID, RequestedBy: userID, + Component: "mysql", TargetType: target.TargetType, TargetID: target.ID, Namespace: input.Namespace, @@ -472,6 +558,11 @@ func (s *DeliveryService) CreateTask(ctx context.Context, userID uint64, isAdmin } return nil, false, err } + if hasCredentialInput { + if err := s.storeDeploymentCredentials(ctx, &task, credentialInput, "pending"); err != nil { + return nil, false, err + } + } _ = s.db.WithContext(ctx).Create(&model.TaskEvent{TaskID: task.ID, ToState: model.TaskPending, Message: "delivery task created"}).Error return &task, false, nil } @@ -620,6 +711,137 @@ func (s *DeliveryService) ListTasks(ctx context.Context, userID uint64, isAdmin return tasks, query.Find(&tasks).Error } +func (s *DeliveryService) ListMySQLServiceLedger(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]MySQLServiceLedgerItem, error) { + if businessLineID == 0 { + return nil, fmt.Errorf("business_line_id is required") + } + if !isAdmin { + var count int64 + if err := s.db.WithContext(ctx).Model(&model.BusinessLineUser{}). + Where("business_line_id = ? AND user_id = ?", businessLineID, userID). + Count(&count).Error; err != nil { + return nil, err + } + if count == 0 { + return nil, fmt.Errorf("user is not authorized for this business line") + } + } + var businessLine model.BusinessLine + if err := s.db.WithContext(ctx).First(&businessLine, "id = ?", businessLineID).Error; err != nil { + return nil, err + } + var instances []model.DeploymentResult + if err := s.db.WithContext(ctx). + Where("business_line_id = ? AND component = ? AND service_type = ? AND status = ?", businessLineID, "mysql", "database", "active"). + Order("created_at DESC"). + Find(&instances).Error; err != nil { + return nil, err + } + taskIDs := make([]string, 0, len(instances)) + for _, instance := range instances { + taskIDs = append(taskIDs, instance.TaskID) + } + taskStatuses := map[string]string{} + if len(taskIDs) > 0 { + var tasks []model.DeliveryTask + if err := s.db.WithContext(ctx).Select("id", "status").Where("id IN ?", taskIDs).Find(&tasks).Error; err != nil { + return nil, err + } + for _, task := range tasks { + taskStatuses[task.ID] = task.Status + } + } + items := make([]MySQLServiceLedgerItem, 0, len(instances)) + for _, instance := range instances { + status := "健康" + statusClass := "ok" + healthy := 1 + if taskStatuses[instance.TaskID] == model.TaskRegisterFailed { + status = "注册异常" + statusClass = "warn" + } + if instance.Host == "" || instance.Port == 0 { + status = "部分异常" + statusClass = "warn" + healthy = 0 + } + items = append(items, MySQLServiceLedgerItem{ + Name: instance.InstanceName, + Datacenter: instance.NodeName, + BusinessTag: businessLine.Name, + Instances: 1, + Healthy: healthy, + Address: fmt.Sprintf("%s:%d", instance.Host, instance.Port), + Status: status, + StatusClass: statusClass, + Version: instance.Version, + Namespace: instance.Namespace, + }) + } + return items, nil +} + +func (s *DeliveryService) storeDeploymentCredentials(ctx context.Context, task *model.DeliveryTask, credentials map[string]string, status string) error { + for key, password := range credentials { + username, host, ok := strings.Cut(key, "@") + if !ok || strings.TrimSpace(username) == "" { + return fmt.Errorf("invalid credential account %q", key) + } + ciphertext, nonce, err := encryptCredential(s.cfg.DeliveryCredentialSecret, password) + if err != nil { + return err + } + item := model.DeploymentCredential{ + TaskID: task.ID, + BusinessLineID: task.BusinessLineID, + Component: task.Component, + InstanceName: task.InstanceName, + Username: strings.TrimSpace(username), + AccountHost: strings.TrimSpace(host), + Ciphertext: ciphertext, + Nonce: nonce, + Status: status, + Source: "user_input", + } + if item.Component == "" { + item.Component = "mysql" + } + if err := s.db.WithContext(ctx).Create(&item).Error; err != nil { + return err + } + } + return nil +} + +func (s *DeliveryService) deploymentCredentialVars(ctx context.Context, taskID string) (map[string]string, error) { + var items []model.DeploymentCredential + if err := s.db.WithContext(ctx). + Where("task_id = ? AND status IN ?", taskID, []string{"pending", "available"}). + Find(&items).Error; err != nil { + return nil, err + } + if len(items) == 0 { + return map[string]string{}, nil + } + values := map[string]string{} + for _, item := range items { + password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, item.Ciphertext, item.Nonce) + if err != nil { + return nil, err + } + switch item.Username + "@" + item.AccountHost { + case "root@localhost": + values["mysql_root_password"] = password + case "xinfra_admin@%": + values["mysql_admin_password"] = password + } + } + if values["mysql_root_password"] == "" || values["mysql_admin_password"] == "" { + return nil, fmt.Errorf("deployment credentials are missing for task %s", taskID) + } + return values, nil +} + func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uint64, isAdmin bool) (*model.DeliveryTask, []model.TaskEvent, error) { query := s.db.WithContext(ctx).Where("id = ?", taskID) if !isAdmin { @@ -636,6 +858,55 @@ func (s *DeliveryService) GetTask(ctx context.Context, taskID string, userID uin return &task, events, 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 { + return nil, err + } + if task.Status != model.TaskFinished && task.Status != model.TaskRegisterFailed { + return nil, fmt.Errorf("task credentials are available only after a successful deployment") + } + var out []DeploymentCredentialView + err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var credentials []model.DeploymentCredential + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("task_id = ? AND status = ?", taskID, "available"). + Order("id ASC"). + Find(&credentials).Error; err != nil { + return err + } + if len(credentials) == 0 { + return gorm.ErrRecordNotFound + } + for _, credential := range credentials { + password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, credential.Ciphertext, credential.Nonce) + if err != nil { + return err + } + out = append(out, DeploymentCredentialView{ + Username: credential.Username, + Host: 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 + }) + if err != nil { + return nil, err + } + return out, nil +} + func (s *DeliveryService) SubscribeTask(taskID string) (<-chan DeliveryTaskSnapshot, func()) { ch := make(chan DeliveryTaskSnapshot, 8) s.streamMu.Lock() @@ -789,7 +1060,9 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT return err } var instancePorts []int - if err := tx.Model(&model.MySQLInstance{}).Where("node_name = ? AND status = ?", host.Name, "active").Pluck("port", &instancePorts).Error; err != nil { + if err := tx.Model(&model.DeploymentResult{}). + Where("component = ? AND service_type = ? AND node_name = ? AND status = ?", "mysql", "database", host.Name, "active"). + Pluck("port", &instancePorts).Error; err != nil { return err } port, portErr := allocatePort(payload.MySQLPort, append(usedPorts, instancePorts...)) @@ -887,6 +1160,14 @@ func randomUUID() string { return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) } +func mustJSON(value any) []byte { + raw, err := json.Marshal(value) + if err != nil { + return []byte(`{}`) + } + return raw +} + func mysqlReady(ctx context.Context, address string) error { dialer := net.Dialer{Timeout: 5 * time.Second} conn, err := dialer.DialContext(ctx, "tcp", address) @@ -952,6 +1233,14 @@ func (s *DeliveryService) CreateExecution(ctx context.Context, taskID, payloadHa return nil, false, err } extraVars := deliveryExtraVars(&task, payload, meta) + credentialVars, err := s.deploymentCredentialVars(ctx, task.ID) + if err != nil { + _ = s.db.WithContext(ctx).Model(&execution).Updates(map[string]any{"status": "failed", "finished_at": time.Now()}) + return nil, false, err + } + for key, value := range credentialVars { + extraVars[key] = value + } extraVars["delivery_callback_url"] = s.deliveryCallbackURL(task.ID) extraVars["delivery_callback_token"] = s.cfg.AWXWebhookToken job, err := s.awx.Launch(ctx, target.AWXTemplateID, AWXLaunchRequest{InventoryID: target.AWXInventoryID, Limit: task.TargetHost, ExtraVars: extraVars}) @@ -1302,11 +1591,31 @@ func (s *DeliveryService) completeTask(ctx context.Context, taskID string) error if err := s.transitionTx(tx, &task, model.TaskRegistering, "AWX succeeded and MySQL health check passed", ""); err != nil { return err } - instance := model.MySQLInstance{TaskID: task.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, Namespace: payload.Namespace, Name: payload.InstanceName, NodeName: task.TargetHost, Host: task.TargetHostIP, Port: task.MySQLPort, Version: payload.MySQLVersion, Status: "active"} - if err := tx.Create(&instance).Error; err != nil { + result := model.DeploymentResult{ + TaskID: task.ID, + BusinessLineID: task.BusinessLineID, + Component: task.Component, + ServiceType: "database", + InstanceName: payload.InstanceName, + Namespace: payload.Namespace, + TargetID: task.TargetID, + NodeName: task.TargetHost, + Host: task.TargetHostIP, + Port: task.MySQLPort, + Version: payload.MySQLVersion, + Status: "active", + Metadata: string(mustJSON(map[string]any{"data_disk": payload.DataDisk})), + } + if result.Component == "" { + result.Component = "mysql" + } + if err := tx.Create(&result).Error; err != nil { return err } - if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: instance.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "active"}).Error; err != nil { + if err := tx.Model(&model.DeploymentCredential{}).Where("task_id = ? AND status = ?", task.ID, "pending").Updates(map[string]any{"deployment_result_id": result.ID, "status": "available"}).Error; err != nil { + return err + } + if err := tx.Create(&model.ResourceUsage{TaskID: task.ID, InstanceID: result.ID, BusinessLineID: task.BusinessLineID, TargetID: task.TargetID, CPUMilli: payload.CPUMilli, MemoryMi: payload.MemoryMi, StorageGi: payload.StorageGi, InstanceCount: 1, Status: "active"}).Error; err != nil { return err } if err := tx.Model(&model.ResourceReservation{}).Where("task_id = ? AND status = ?", task.ID, "reserved").Update("status", "consumed").Error; err != nil { @@ -1359,11 +1668,11 @@ func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) er if s.cfg.CloudDMRegisterURL == "" { return nil } - var instance model.MySQLInstance - if err := s.db.WithContext(ctx).Where("task_id = ?", taskID).First(&instance).Error; err != nil { + var instance model.DeploymentResult + if err := s.db.WithContext(ctx).Where("task_id = ? AND component = ? AND service_type = ?", taskID, "mysql", "database").First(&instance).Error; err != nil { return err } - body := map[string]any{"name": instance.Name, "host": instance.Host, "port": instance.Port, "username": "root", "database_type": "mysql"} + body := map[string]any{"name": instance.InstanceName, "host": instance.Host, "port": instance.Port, "username": "root", "database_type": "mysql"} raw, _ := json.Marshal(body) req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMRegisterURL, bytes.NewReader(raw)) if err != nil { @@ -1509,7 +1818,7 @@ func (s *DeliveryService) AcknowledgeRollbackRelease(ctx context.Context, taskID if task.Status != model.TaskRollbackFailed { return fmt.Errorf("task %s is in state %q and cannot acknowledge rollback release", taskID, task.Status) } - if err := tx.Model(&model.MySQLInstance{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "rollback_acknowledged", "updated_at": now}).Error; err != nil { + if err := tx.Model(&model.DeploymentResult{}).Where("task_id = ? AND component = ? AND service_type = ? AND status = ?", taskID, "mysql", "database", "active").Updates(map[string]any{"status": "rollback_acknowledged", "updated_at": now}).Error; err != nil { return err } if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil { @@ -1627,7 +1936,10 @@ func (s *DeliveryService) completeRollback(ctx context.Context, taskID string) e if task.Status != model.TaskRollingBack { return fmt.Errorf("task %s is in state %q, cannot complete rollback", taskID, task.Status) } - if err := tx.Model(&model.MySQLInstance{}).Where("task_id = ?", taskID).Updates(map[string]any{"status": "rolled_back", "updated_at": now}).Error; err != nil { + if err := tx.Model(&model.DeploymentResult{}).Where("task_id = ?", taskID).Updates(map[string]any{"status": "rolled_back", "updated_at": now}).Error; err != nil { + return err + } + if err := tx.Unscoped().Where("task_id = ?", taskID).Delete(&model.DeploymentCredential{}).Error; err != nil { return err } if err := tx.Model(&model.ResourceUsage{}).Where("task_id = ? AND status = ?", taskID, "active").Updates(map[string]any{"status": "released", "released_at": now, "updated_at": now}).Error; err != nil { diff --git a/server/internal/service/machine.go b/server/internal/service/machine.go new file mode 100644 index 0000000..b104ddf --- /dev/null +++ b/server/internal/service/machine.go @@ -0,0 +1,675 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" + + "github.com/1024XEngineer/xinfra/server/internal/config" + "github.com/1024XEngineer/xinfra/server/internal/model" + + "gorm.io/gorm" +) + +const ( + machineSyncSource = "sina" + machineClassName = "zion_server" + sinaLoginPath = "/sinai/v1/login" + sinaCILotPath = "/sinai/v1/3rd/ci/lot" + sinaSyncInterval = 30 * time.Minute + sinaRequestTimeout = 10 * time.Second + sinaPageSize = 500 +) + +type MachineService struct { + cfg config.Config + db *gorm.DB + httpClient *http.Client + mu sync.Mutex + running bool +} + +type MachineListQuery struct { + Page int + Size int + Hostname string + AssetNumber string + Type string + Location string + IP string + Spec string + BusinessLine string + Source string + Status string +} + +type MachineResourceItem struct { + Hostname string `json:"hostname"` + AssetNumber string `json:"assetNumber"` + Type string `json:"type"` + Location string `json:"location"` + IP string `json:"ip"` + Spec string `json:"spec"` + BusinessLine string `json:"businessLine"` + Source string `json:"source"` + Status string `json:"status"` +} + +type MachineResourceList struct { + Total int64 `json:"total"` + Items []MachineResourceItem `json:"items"` +} + +type MachineOverview struct { + Total int64 `json:"total"` + Physical int64 `json:"physical"` + Virtual int64 `json:"virtual"` + CMDB MachineSourceStat `json:"cmdb"` + Aliyun MachineCloudStat `json:"aliyun"` + AWSQiniu MachineAWSQiniuStat `json:"awsQiniu"` + LastSync MachineLastSyncOverview `json:"lastSync"` +} + +type MachineSourceStat struct { + Total int64 `json:"total"` + Physical int64 `json:"physical"` + Virtual int64 `json:"virtual"` +} + +type MachineCloudStat struct { + Total int64 `json:"total"` + Label string `json:"label"` +} + +type MachineAWSQiniuStat struct { + Total int64 `json:"total"` + AWS int64 `json:"aws"` + Qiniu int64 `json:"qiniu"` +} + +type MachineLastSyncOverview struct { + Status string `json:"status"` + Time string `json:"time"` + Created int64 `json:"created"` + Error string `json:"error,omitempty"` +} + +func NewMachineService(cfg config.Config, db *gorm.DB) *MachineService { + return &MachineService{ + cfg: cfg, + db: db, + httpClient: &http.Client{ + Timeout: sinaRequestTimeout, + }, + } +} + +func (s *MachineService) Run(ctx context.Context) { + go func() { + _, _ = s.SyncNow(ctx) + ticker := time.NewTicker(sinaSyncInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + _, _ = s.SyncNow(ctx) + } + } + }() +} + +func (s *MachineService) SyncNow(ctx context.Context) (*model.MachineSyncState, error) { + if err := s.acquireSync(); err != nil { + state, _ := s.syncState(ctx) + return state, err + } + defer s.releaseSync() + + started := time.Now() + if err := s.startSync(ctx, started); err != nil { + return nil, err + } + token, err := s.loginSina(ctx) + if err != nil { + _ = s.finishSync(ctx, 0, 0, 0, 0, err) + state, _ := s.syncState(ctx) + return state, err + } + + page := 1 + var total, fetched, created, updated int64 + for { + resp, err := s.fetchSinaPage(ctx, token, page, sinaPageSize) + if err != nil { + _ = s.finishSync(ctx, total, fetched, created, updated, err) + state, _ := s.syncState(ctx) + return state, err + } + if page == 1 { + total = resp.Result.Count + } + if len(resp.Result.Items) == 0 { + break + } + for _, item := range resp.Result.Items { + isCreated, err := s.upsertMachine(ctx, item, started) + if err != nil { + _ = s.finishSync(ctx, total, fetched, created, updated, err) + state, _ := s.syncState(ctx) + return state, err + } + fetched++ + if isCreated { + created++ + } else { + updated++ + } + } + if total > 0 && fetched >= total { + break + } + if len(resp.Result.Items) < sinaPageSize { + break + } + page++ + } + + if err := s.finishSync(ctx, total, fetched, created, updated, nil); err != nil { + return nil, err + } + return s.syncState(ctx) +} + +func (s *MachineService) Overview(ctx context.Context) (*MachineOverview, error) { + var total, physical int64 + if err := s.db.WithContext(ctx).Model(&model.MachineResource{}).Count(&total).Error; err != nil { + return nil, err + } + if err := s.db.WithContext(ctx).Model(&model.MachineResource{}).Where("resource_type = ?", "physical").Count(&physical).Error; err != nil { + return nil, err + } + + sourceCounts, err := s.sourceCounts(ctx) + if err != nil { + return nil, err + } + cloudSources := []string{"aliyun", "ali", "alicloud", "aws", "qiniu"} + cmdbTotal, err := s.countExcludingSources(ctx, cloudSources) + if err != nil { + return nil, err + } + cmdbPhysical, err := s.countExcludingSourcesAndType(ctx, cloudSources, "physical") + if err != nil { + return nil, err + } + + state, err := s.syncState(ctx) + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, err + } + + return &MachineOverview{ + Total: total, + Physical: physical, + Virtual: total - physical, + CMDB: MachineSourceStat{ + Total: cmdbTotal, + Physical: cmdbPhysical, + Virtual: cmdbTotal - cmdbPhysical, + }, + Aliyun: MachineCloudStat{ + Total: sourceCounts["aliyun"], + Label: "ECS · 增量同步", + }, + AWSQiniu: MachineAWSQiniuStat{ + Total: sourceCounts["aws"] + sourceCounts["qiniu"], + AWS: sourceCounts["aws"], + Qiniu: sourceCounts["qiniu"], + }, + LastSync: mapLastSync(state), + }, nil +} + +func (s *MachineService) List(ctx context.Context, query MachineListQuery) (*MachineResourceList, error) { + if query.Page <= 0 { + query.Page = 1 + } + if query.Size <= 0 { + query.Size = 20 + } + if query.Size > 200 { + query.Size = 200 + } + + db := s.db.WithContext(ctx).Model(&model.MachineResource{}) + db = applyMachineFilters(db, query) + + var total int64 + if err := db.Count(&total).Error; err != nil { + return nil, err + } + + var rows []model.MachineResource + if err := db.Order("updated_at DESC").Offset((query.Page - 1) * query.Size).Limit(query.Size).Find(&rows).Error; err != nil { + return nil, err + } + + items := make([]MachineResourceItem, 0, len(rows)) + for _, row := range rows { + items = append(items, mapMachineResource(row)) + } + return &MachineResourceList{Total: total, Items: items}, nil +} + +func (s *MachineService) acquireSync() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.running { + return errors.New("machine sync is already running") + } + s.running = true + return nil +} + +func (s *MachineService) releaseSync() { + s.mu.Lock() + defer s.mu.Unlock() + s.running = false +} + +func (s *MachineService) startSync(ctx context.Context, started time.Time) error { + state := model.MachineSyncState{ + Source: machineSyncSource, + Status: "running", + LastStartedAt: &started, + LastError: "", + } + return s.db.WithContext(ctx).Where("source = ?", machineSyncSource).Assign(state).FirstOrCreate(&state).Error +} + +func (s *MachineService) finishSync(ctx context.Context, total, fetched, created, updated int64, syncErr error) error { + now := time.Now() + status := "success" + lastErr := "" + if syncErr != nil { + status = "failed" + lastErr = syncErr.Error() + } + return s.db.WithContext(ctx).Model(&model.MachineSyncState{}). + Where("source = ?", machineSyncSource). + Updates(map[string]interface{}{ + "status": status, + "last_finished_at": &now, + "last_error": lastErr, + "total": total, + "fetched": fetched, + "created_count": created, + "updated_count": updated, + }).Error +} + +func (s *MachineService) syncState(ctx context.Context) (*model.MachineSyncState, error) { + var state model.MachineSyncState + err := s.db.WithContext(ctx).Where("source = ?", machineSyncSource).First(&state).Error + return &state, err +} + +type sinaPageResp struct { + Success bool `json:"success"` + Result struct { + Items []map[string]interface{} `json:"items"` + Count int64 `json:"count"` + } `json:"result"` + Message string `json:"message"` +} + +type sinaLoginResp struct { + Success bool `json:"success"` + Result map[string]interface{} `json:"result"` + Data map[string]interface{} `json:"data"` + Token string `json:"token"` + Message string `json:"message"` +} + +func (s *MachineService) loginSina(ctx context.Context) (string, error) { + username := strings.TrimSpace(s.cfg.SINAUsername) + password := strings.TrimSpace(s.cfg.SINAPassword) + if username == "" || password == "" { + return "", errors.New("SINA_USERNAME or SINA_PASSWORD is not configured") + } + + payload, err := json.Marshal(map[string]string{ + "username": username, + "password": password, + }) + if err != nil { + return "", err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.SINABaseURL+sinaLoginPath, bytes.NewReader(payload)) + if err != nil { + return "", err + } + req.Header.Set("content-type", "application/json") + + resp, err := s.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("sina login failed: status=%d body=%s", resp.StatusCode, string(respBody)) + } + + var parsed sinaLoginResp + if err := json.Unmarshal(respBody, &parsed); err != nil { + return "", err + } + if !parsed.Success { + return "", fmt.Errorf("sina login failed: %s", parsed.Message) + } + token := firstNonEmpty( + parsed.Token, + machineStringValue(parsed.Result["token"]), + machineStringValue(parsed.Result["access_token"]), + machineStringValue(parsed.Data["token"]), + machineStringValue(parsed.Data["access_token"]), + ) + if token == "" { + return "", errors.New("sina login response missing token") + } + return token, nil +} + +func (s *MachineService) fetchSinaPage(ctx context.Context, token string, page, size int) (*sinaPageResp, error) { + body := map[string]interface{}{ + "className": machineClassName, + "page": page, + "size": size, + "field": map[string]interface{}{}, + } + payload, err := json.Marshal(body) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.SINABaseURL+sinaCILotPath, bytes.NewReader(payload)) + if err != nil { + return nil, err + } + req.Header.Set("content-type", "application/json") + req.Header.Set("Authorization", token) + + resp, err := s.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("sina request failed: status=%d body=%s", resp.StatusCode, string(respBody)) + } + var parsed sinaPageResp + if err := json.Unmarshal(respBody, &parsed); err != nil { + return nil, err + } + if !parsed.Success { + return nil, fmt.Errorf("sina request failed: %s", parsed.Message) + } + return &parsed, nil +} + +func (s *MachineService) upsertMachine(ctx context.Context, item map[string]interface{}, syncedAt time.Time) (bool, error) { + sinaID := machineInt64Value(item["id"]) + if sinaID == 0 { + return false, nil + } + raw, _ := json.Marshal(item) + next := model.MachineResource{ + SinaID: sinaID, + CIID: machineStringValue(item["ciId"]), + Hostname: machineStringValue(item["hostname"]), + Name: machineStringValue(item["name"]), + AssetNumber: machineStringValue(item["asset_number"]), + IDCNumber: machineStringValue(item["idc_number"]), + ResourceType: machineStringValue(item["type"]), + LocationName: machineStringValue(item["location_name"]), + LocationAlias: machineStringValue(item["location_alias"]), + IntranetIP: machineStringValue(item["intranet_ip"]), + Spec: machineStringValue(item["spec"]), + BusinessLine: machineStringValue(item["org_name"]), + Source: machineStringValue(item["source"]), + Status: machineStringValue(item["status"]), + Brand: machineStringValue(item["brand"]), + Model: machineStringValue(item["model"]), + SerialNumber: machineStringValue(item["serial_number"]), + PowerStatus: machineStringValue(item["power_status"]), + CPUCores: machineInt64Value(item["cpu_cores"]), + CPUName: machineStringValue(item["cpu_name"]), + Memory: machineStringValue(item["memory"]), + HardCapacity: machineStringValue(item["hard_capacity"]), + SSDCapacity: machineStringValue(item["ssd_capacity"]), + OOBIP: machineStringValue(item["oob_ip"]), + OSFamily: machineStringValue(item["os_family"]), + OSVersion: machineStringValue(item["os_version"]), + CollectTime: parseSinaTime(machineStringValue(item["collect_time"])), + SinaUpdatedAt: parseSinaTime(machineStringValue(item["updated"])), + RawJSON: string(raw), + LastSyncedAt: syncedAt, + } + + var current model.MachineResource + err := s.db.WithContext(ctx).Where("sina_id = ?", sinaID).First(¤t).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return true, s.db.WithContext(ctx).Create(&next).Error + } + if err != nil { + return false, err + } + next.ID = current.ID + next.CreatedAt = current.CreatedAt + return false, s.db.WithContext(ctx).Save(&next).Error +} + +func applyMachineFilters(db *gorm.DB, query MachineListQuery) *gorm.DB { + switch query.Type { + case "physical": + db = db.Where("resource_type = ?", "physical") + case "vm", "virtual": + db = db.Where("resource_type <> ?", "physical") + } + + switch query.Source { + case "cmdb": + db = db.Where("source NOT IN ?", []string{"aliyun", "ali", "alicloud", "aws", "qiniu"}) + case "aliyun": + db = db.Where("source IN ?", []string{"aliyun", "ali", "alicloud"}) + case "aws": + db = db.Where("source = ?", "aws") + case "qiniu": + db = db.Where("source = ?", "qiniu") + } + + if query.Status != "" { + db = db.Where("status = ?", query.Status) + } + if query.Hostname != "" { + like := "%" + query.Hostname + "%" + db = db.Where("hostname LIKE ? OR name LIKE ?", like, like) + } + if query.AssetNumber != "" { + like := "%" + query.AssetNumber + "%" + db = db.Where("asset_number LIKE ? OR idc_number LIKE ?", like, like) + } + if query.Location != "" { + like := "%" + query.Location + "%" + db = db.Where("location_name LIKE ? OR location_alias LIKE ?", like, like) + } + if query.IP != "" { + db = db.Where("intranet_ip LIKE ?", "%"+query.IP+"%") + } + if query.Spec != "" { + db = db.Where("spec LIKE ?", "%"+query.Spec+"%") + } + if query.BusinessLine != "" { + db = db.Where("business_line LIKE ?", "%"+query.BusinessLine+"%") + } + return db +} + +func (s *MachineService) sourceCounts(ctx context.Context) (map[string]int64, error) { + var rows []struct { + Source string + Count int64 + } + if err := s.db.WithContext(ctx).Model(&model.MachineResource{}).Select("source, count(*) as count").Group("source").Scan(&rows).Error; err != nil { + return nil, err + } + out := map[string]int64{} + for _, row := range rows { + key := classifyMachineSource(row.Source) + out[key] += row.Count + } + return out, nil +} + +func (s *MachineService) countExcludingSources(ctx context.Context, sources []string) (int64, error) { + var count int64 + err := s.db.WithContext(ctx).Model(&model.MachineResource{}).Where("source NOT IN ?", sources).Count(&count).Error + return count, err +} + +func (s *MachineService) countExcludingSourcesAndType(ctx context.Context, sources []string, resourceType string) (int64, error) { + var count int64 + err := s.db.WithContext(ctx).Model(&model.MachineResource{}).Where("source NOT IN ? AND resource_type = ?", sources, resourceType).Count(&count).Error + return count, err +} + +func mapMachineResource(row model.MachineResource) MachineResourceItem { + return MachineResourceItem{ + Hostname: firstNonEmpty(row.Hostname, row.Name), + AssetNumber: firstNonEmpty(row.AssetNumber, row.IDCNumber, "—"), + Type: displayMachineType(row.ResourceType), + Location: firstNonEmpty(row.LocationName, row.LocationAlias), + IP: row.IntranetIP, + Spec: row.Spec, + BusinessLine: row.BusinessLine, + Source: displayMachineSource(row.Source), + Status: row.Status, + } +} + +func mapLastSync(state *model.MachineSyncState) MachineLastSyncOverview { + if state == nil || state.ID == 0 || state.LastFinishedAt == nil { + return MachineLastSyncOverview{Status: "unknown", Time: "—"} + } + status := "normal" + if state.Status == "failed" { + status = "failed" + } + return MachineLastSyncOverview{ + Status: status, + Time: state.LastFinishedAt.Format("15:04:05"), + Created: state.CreatedCount, + Error: state.LastError, + } +} + +func classifyMachineSource(source string) string { + switch strings.ToLower(strings.TrimSpace(source)) { + case "aliyun", "ali", "alicloud": + return "aliyun" + case "aws": + return "aws" + case "qiniu": + return "qiniu" + default: + return "cmdb" + } +} + +func displayMachineSource(source string) string { + switch classifyMachineSource(source) { + case "aliyun": + return "阿里云同步" + case "aws": + return "AWS 同步" + case "qiniu": + return "七牛同步" + default: + return "SINA CMDB" + } +} + +func displayMachineType(value string) string { + if strings.EqualFold(value, "physical") { + return "物理机" + } + return "虚机" +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + +func machineStringValue(value interface{}) string { + switch v := value.(type) { + case string: + return strings.TrimSpace(v) + case fmt.Stringer: + return strings.TrimSpace(v.String()) + case nil: + return "" + default: + return strings.TrimSpace(fmt.Sprint(v)) + } +} + +func machineInt64Value(value interface{}) int64 { + switch v := value.(type) { + case int64: + return v + case int: + return int64(v) + case float64: + return int64(v) + case json.Number: + n, _ := v.Int64() + return n + default: + return 0 + } +} + +func parseSinaTime(value string) *time.Time { + value = strings.TrimSpace(value) + if value == "" || strings.HasPrefix(value, "0001-") { + return nil + } + t, err := time.ParseInLocation("2006-01-02 15:04:05", value, time.Local) + if err != nil { + return nil + } + return &t +}