feat: wire delivery result and infra data backends

This commit is contained in:
mac
2026-07-28 17:56:00 +08:00
parent 897670b721
commit f7be87c7d9
14 changed files with 1468 additions and 34 deletions
+3 -3
View File
@@ -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('') }}"
+3 -3
View File
@@ -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
+29
View File
@@ -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=
+8
View File
@@ -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", ""),
}
}
+4 -1
View File
@@ -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{},
)
}
@@ -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 {
+41
View File
@@ -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 返回指定任务的详细信息及事件流
+69
View File
@@ -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
}
+30 -9
View File
@@ -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"`
+54
View File
@@ -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"`
}
+9
View File
@@ -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)
}
}
+214 -9
View File
@@ -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")
+321 -9
View File
@@ -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 {
+675
View File
@@ -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(&current).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
}