701 lines
23 KiB
Go
701 lines
23 KiB
Go
package server
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/jackc/pgx/v5/pgconn"
|
||
"github.com/jmoiron/sqlx"
|
||
"log"
|
||
"net"
|
||
"strings"
|
||
"time"
|
||
"vdi-cloud/platform/internal/model"
|
||
"vdi-cloud/platform/internal/store"
|
||
"vdi-cloud/platform/pkg/proxmox"
|
||
)
|
||
|
||
func (s *Server) overview(c *gin.Context) {
|
||
var counts struct {
|
||
Terminals int `db:"terminals"`
|
||
Online int `db:"online"`
|
||
Desktops int `db:"desktops"`
|
||
Running int `db:"running"`
|
||
Sessions int `db:"sessions"`
|
||
Failed int `db:"failed"`
|
||
}
|
||
err := s.DB.GetContext(c.Request.Context(), &counts, `SELECT (SELECT count(*) FROM terminals) terminals,(SELECT count(*) FROM terminals WHERE enabled AND last_seen>now()-interval '90 seconds') online,(SELECT count(*) FROM desktops) desktops,(SELECT count(*) FROM desktops WHERE power_state='running') running,(SELECT count(*) FROM terminals WHERE enabled AND session_state='connected' AND session_desktop_id IS NOT NULL AND last_seen>now()-interval '90 seconds') sessions,(SELECT count(*) FROM tasks WHERE status IN ('failed','timeout') AND created_at>now()-interval '24 hours') failed`)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
c.JSON(200, gin.H{"terminalCount": counts.Terminals, "onlineTerminals": counts.Online, "desktopCount": counts.Desktops, "runningDesktops": counts.Running, "activeSessions": counts.Sessions, "failedTasks": counts.Failed, "mode": s.Config.Mode})
|
||
}
|
||
func (s *Server) settings(c *gin.Context) {
|
||
pve := s.currentPVE()
|
||
provider := "unconfigured"
|
||
if s.Config.Mode == "demo" {
|
||
provider = "demo"
|
||
}
|
||
if pve != nil {
|
||
provider = "proxmox"
|
||
}
|
||
c.JSON(200, gin.H{"mode": s.Config.Mode, "computeProvider": provider, "pveConfigured": pve != nil, "terminalTransport": "https-poll", "capabilities": []string{"diagnostics-center", "admin-mfa", "administrator-sessions", "task-history", "audit-query-export", "terminal-metric-alerts", "terminal-system-metrics", "gpu-inventory", "desktop-batches", "desktop-snapshots", "pve-connection", "desktop-hardware", "login-rate-limit", "csv-imports", "session-history", "monitor-alerts", "administrators", "password", "auth", "resources", "desktops", "templates", "pools", "clone", "desktop-connection", "terminals", "terminal-lifecycle", "terminal-policies", "users", "batch-bindings", "bindings", "tasks", "audit"}})
|
||
}
|
||
func (s *Server) resources(c *gin.Context) {
|
||
items := []model.Resource{}
|
||
if err := s.DB.SelectContext(c.Request.Context(), &items, "SELECT * FROM resources ORDER BY name"); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
list(c, items)
|
||
}
|
||
func (s *Server) desktops(c *gin.Context) {
|
||
items := []model.Desktop{}
|
||
q := "%" + c.Query("keyword") + "%"
|
||
err := s.DB.SelectContext(c.Request.Context(), &items, `SELECT * FROM desktops WHERE (name ILIKE $1 OR vmid::text ILIKE $1) AND ($2='' OR power_state=$2) ORDER BY name`, q, c.Query("status"))
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
list(c, items)
|
||
}
|
||
func (s *Server) desktop(c *gin.Context) {
|
||
var d model.Desktop
|
||
if err := s.DB.GetContext(c.Request.Context(), &d, "SELECT * FROM desktops WHERE id=$1", c.Param("id")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
c.JSON(200, d)
|
||
}
|
||
func (s *Server) terminals(c *gin.Context) {
|
||
items := []model.Terminal{}
|
||
err := s.DB.SelectContext(c.Request.Context(), &items, terminalListSQL+` WHERE (name ILIKE $1 OR sn ILIKE $1) ORDER BY name`, "%"+c.Query("keyword")+"%")
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
list(c, items)
|
||
}
|
||
func (s *Server) users(c *gin.Context) {
|
||
items, err := store.Users(c.Request.Context(), s.DB)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
list(c, items)
|
||
}
|
||
func (s *Server) createUser(c *gin.Context) {
|
||
var in struct {
|
||
Username string `json:"username" binding:"required,min=2,max=80"`
|
||
DisplayName string `json:"displayName" binding:"required,max=80"`
|
||
}
|
||
if !s.decode(c, &in) {
|
||
return
|
||
}
|
||
in.Username = strings.TrimSpace(in.Username)
|
||
in.DisplayName = strings.TrimSpace(in.DisplayName)
|
||
if in.Username == "" || in.DisplayName == "" {
|
||
s.fail(c, 400, "invalid_request", "账号和姓名不能为空")
|
||
return
|
||
}
|
||
id := store.ID("user_")
|
||
_, err := s.DB.ExecContext(c.Request.Context(), "INSERT INTO users(id,username,display_name) VALUES($1,$2,$3)", id, in.Username, in.DisplayName)
|
||
if err != nil {
|
||
if conflict(err) {
|
||
s.fail(c, 409, "conflict", "该账号已存在")
|
||
} else {
|
||
s.dbError(c, err)
|
||
}
|
||
return
|
||
}
|
||
s.log(c, "create_user", id, "success")
|
||
c.JSON(201, model.User{ID: id, Username: in.Username, DisplayName: in.DisplayName, Enabled: true, DesktopIDs: []string{}})
|
||
}
|
||
func conflict(err error) bool {
|
||
var e *pgconn.PgError
|
||
return errors.As(err, &e) && (e.Code == "23505" || e.Code == "23503")
|
||
}
|
||
func (s *Server) bindings(c *gin.Context) {
|
||
var in struct {
|
||
TerminalID *string `json:"terminalId"`
|
||
DesktopIDs []string `json:"desktopIds" binding:"max=100"`
|
||
Enabled *bool `json:"enabled"`
|
||
}
|
||
if !s.decode(c, &in) {
|
||
return
|
||
}
|
||
if in.TerminalID != nil && *in.TerminalID == "" {
|
||
in.TerminalID = nil
|
||
}
|
||
ctx := c.Request.Context()
|
||
tx, err := s.DB.BeginTxx(ctx, nil)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
defer tx.Rollback()
|
||
if !s.lockAuthorization(c, tx) {
|
||
return
|
||
}
|
||
var user model.User
|
||
if err = tx.GetContext(ctx, &user, "SELECT * FROM users WHERE id=$1 FOR UPDATE", c.Param("id")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if in.Enabled != nil {
|
||
user.Enabled = *in.Enabled
|
||
}
|
||
_, err = tx.ExecContext(ctx, "UPDATE users SET terminal_id=$1,enabled=$2 WHERE id=$3", in.TerminalID, user.Enabled, user.ID)
|
||
if err != nil {
|
||
if conflict(err) {
|
||
s.fail(c, 409, "binding_conflict", "终端不存在或已关联其他用户")
|
||
} else {
|
||
s.dbError(c, err)
|
||
}
|
||
return
|
||
}
|
||
if _, err = tx.ExecContext(ctx, "DELETE FROM bindings WHERE user_id=$1", user.ID); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
seen := map[string]bool{}
|
||
for _, id := range in.DesktopIDs {
|
||
if seen[id] {
|
||
continue
|
||
}
|
||
seen[id] = true
|
||
if _, err = tx.ExecContext(ctx, "INSERT INTO bindings VALUES($1,$2)", user.ID, id); err != nil {
|
||
if conflict(err) {
|
||
s.fail(c, 400, "invalid_desktop", "所选桌面不存在")
|
||
} else {
|
||
s.dbError(c, err)
|
||
}
|
||
return
|
||
}
|
||
}
|
||
if _, err = tx.ExecContext(ctx, "DELETE FROM connection_tickets WHERE consumed_at IS NULL AND (terminal_id=$1 OR terminal_id=$2)", user.TerminalID, in.TerminalID); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO audit_logs VALUES($1,$2,'update_bindings',$3,'success',$4,now())", store.ID("audit_"), s.admin(c).Username, user.ID, c.GetString("requestId"))
|
||
if err == nil {
|
||
err = tx.Commit()
|
||
}
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
user.TerminalID = in.TerminalID
|
||
user.DesktopIDs = in.DesktopIDs
|
||
if user.DesktopIDs == nil {
|
||
user.DesktopIDs = []string{}
|
||
}
|
||
c.JSON(200, user)
|
||
}
|
||
func (s *Server) task(c *gin.Context) {
|
||
var t model.Task
|
||
if err := s.DB.GetContext(c.Request.Context(), &t, "SELECT * FROM tasks WHERE id=$1", c.Param("id")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
c.JSON(200, t)
|
||
}
|
||
func (s *Server) discoverVMs(c *gin.Context) {
|
||
pve := s.currentPVE()
|
||
if pve == nil {
|
||
s.fail(c, 503, "pve_unconfigured", "尚未配置 Proxmox 接入，演示环境不发现真实 VM")
|
||
return
|
||
}
|
||
items, err := pve.Resources(c.Request.Context())
|
||
if err != nil {
|
||
s.fail(c, 502, "pve_unavailable", "Proxmox 查询失败，请检查连接、证书和访问权限")
|
||
return
|
||
}
|
||
vms := []gin.H{}
|
||
for _, r := range items {
|
||
if r.Type == "qemu" && r.Template == 0 {
|
||
vms = append(vms, gin.H{"vmid": r.VMID, "node": r.Node, "name": r.Name, "status": r.Status})
|
||
}
|
||
}
|
||
list(c, vms)
|
||
}
|
||
func (s *Server) syncResources(c *gin.Context) {
|
||
pve := s.currentPVE()
|
||
if pve == nil {
|
||
s.fail(c, 503, "pve_unconfigured", "请通过服务端环境变量配置 Proxmox 接入")
|
||
return
|
||
}
|
||
s.submitTask(c, "resource", "sync", "cluster", "Proxmox 集群")
|
||
}
|
||
func (s *Server) importDesktop(c *gin.Context) {
|
||
pve := s.currentPVE()
|
||
var in struct {
|
||
Name string `json:"name" binding:"required,max=80"`
|
||
OS string `json:"os" binding:"required,oneof=Windows Ubuntu 麒麟"`
|
||
Node string `json:"node" binding:"required,max=100"`
|
||
VMID int `json:"vmid" binding:"min=100"`
|
||
Endpoint string `json:"endpoint" binding:"max=253"`
|
||
Width int `json:"width" binding:"omitempty,min=640,max=7680"`
|
||
Height int `json:"height" binding:"omitempty,min=480,max=4320"`
|
||
FPS int `json:"fps" binding:"omitempty,min=10,max=120"`
|
||
}
|
||
if !s.decode(c, &in) {
|
||
return
|
||
}
|
||
if pve == nil {
|
||
s.fail(c, 503, "pve_unconfigured", "真实 VM 纳管需要先配置 Proxmox")
|
||
return
|
||
}
|
||
if in.Endpoint != "" && net.ParseIP(in.Endpoint) == nil {
|
||
for _, part := range strings.Split(in.Endpoint, ".") {
|
||
if part == "" || strings.ContainsAny(part, "/:@ \\") {
|
||
s.fail(c, 400, "invalid_endpoint", "连接地址应为 IP 或主机名")
|
||
return
|
||
}
|
||
}
|
||
}
|
||
power, err := pve.Power(c.Request.Context(), in.Node, in.VMID)
|
||
if err != nil {
|
||
s.fail(c, 502, "pve_unavailable", "无法验证所选 VM，请检查节点和 VMID")
|
||
return
|
||
}
|
||
id := store.ID("desk_")
|
||
if in.Width == 0 {
|
||
in.Width = 2560
|
||
}
|
||
if in.Height == 0 {
|
||
in.Height = 1440
|
||
}
|
||
if in.FPS == 0 {
|
||
in.FPS = 30
|
||
}
|
||
_, err = s.DB.ExecContext(c.Request.Context(), `INSERT INTO desktops(id,name,os,node,vmid,power_state,provider,endpoint,width,height,fps) VALUES($1,$2,$3,$4,$5,$6,'proxmox',$7,$8,$9,$10)`, id, in.Name, in.OS, in.Node, in.VMID, power, in.Endpoint, in.Width, in.Height, in.FPS)
|
||
if err != nil {
|
||
if conflict(err) {
|
||
s.fail(c, 409, "already_managed", "该 VM 已纳管")
|
||
} else {
|
||
s.dbError(c, err)
|
||
}
|
||
return
|
||
}
|
||
s.log(c, "import_desktop", id, "success")
|
||
c.Params = append(c.Params, gin.Param{Key: "id", Value: id})
|
||
s.desktop(c)
|
||
}
|
||
func (s *Server) desktopAction(c *gin.Context) {
|
||
pve := s.currentPVE()
|
||
var in struct {
|
||
Action string `json:"action" binding:"required,oneof=start shutdown reboot"`
|
||
}
|
||
if !s.decode(c, &in) {
|
||
return
|
||
}
|
||
var d model.Desktop
|
||
if err := s.DB.GetContext(c.Request.Context(), &d, "SELECT * FROM desktops WHERE id=$1", c.Param("id")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if d.Provider == "demo" && s.Config.Mode != "demo" {
|
||
s.fail(c, 409, "demo_disabled", "正式模式不执行演示桌面任务")
|
||
return
|
||
}
|
||
if d.Provider == "proxmox" && pve == nil {
|
||
s.fail(c, 503, "pve_unconfigured", "Proxmox 尚未配置")
|
||
return
|
||
}
|
||
if s.exclusiveGPU(d) && in.Action != "shutdown" {
|
||
s.fail(c, 409, "exclusive_gpu_power", "此桌面共用显卡，请先退出当前云桌面，再从终端选择页启动")
|
||
return
|
||
}
|
||
unresolved, err := maintenanceBlocked(c.Request.Context(), s.DB, d.ID, "", false)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if unresolved {
|
||
s.fail(c, 409, "hardware_unconfirmed", "请先核对结果未确认的桌面维护任务")
|
||
return
|
||
}
|
||
s.submitTask(c, "desktop", in.Action, d.ID, d.Name)
|
||
}
|
||
func (s *Server) terminalCommand(c *gin.Context) {
|
||
var in struct {
|
||
Action string `json:"action" binding:"required,oneof=reboot reset-session collect-logs"`
|
||
}
|
||
if !s.decode(c, &in) {
|
||
return
|
||
}
|
||
var name string
|
||
if err := s.DB.GetContext(c.Request.Context(), &name, "SELECT name FROM terminals WHERE id=$1", c.Param("id")); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
s.submitTask(c, "terminal", in.Action, c.Param("id"), name)
|
||
}
|
||
func (s *Server) submitTask(c *gin.Context, kind, action, targetID, targetName string) {
|
||
key := c.GetHeader("Idempotency-Key")
|
||
if key == "" || len(key) > 100 {
|
||
s.fail(c, 400, "idempotency_required", "请提供不超过 100 字符的 Idempotency-Key")
|
||
return
|
||
}
|
||
actor := s.admin(c).Username
|
||
ctx := c.Request.Context()
|
||
var existing model.Task
|
||
err := s.DB.GetContext(ctx, &existing, "SELECT * FROM tasks WHERE actor=$1 AND idempotency_key=$2", actor, key)
|
||
if err == nil {
|
||
if existing.Kind != kind || existing.Action != action || existing.TargetID != targetID {
|
||
s.fail(c, 409, "idempotency_conflict", "相同请求标识不能用于不同操作")
|
||
return
|
||
}
|
||
c.JSON(202, existing)
|
||
return
|
||
}
|
||
if !errors.Is(err, sql.ErrNoRows) {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
tx, err := s.DB.BeginTxx(ctx, nil)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
defer tx.Rollback()
|
||
if kind == "desktop" {
|
||
var desktop model.Desktop
|
||
if err = tx.GetContext(ctx, &desktop, "SELECT * FROM desktops WHERE id=$1 FOR UPDATE", targetID); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if err = s.exclusiveGPUMaintenance(ctx, tx, desktop); err != nil {
|
||
s.fail(c, 409, "gpu_switch_busy", err.Error())
|
||
return
|
||
}
|
||
unresolved, err := maintenanceBlocked(ctx, tx, targetID, "", false)
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if unresolved {
|
||
s.fail(c, 409, "hardware_unconfirmed", "请先核对未确认的桌面维护任务")
|
||
return
|
||
}
|
||
}
|
||
if kind == "terminal" {
|
||
var enabled bool
|
||
if err = tx.GetContext(ctx, &enabled, "SELECT enabled FROM terminals WHERE id=$1 FOR UPDATE", targetID); err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
if !enabled {
|
||
s.fail(c, 409, "terminal_disabled", "终端已停用，请重新激活后操作")
|
||
return
|
||
}
|
||
}
|
||
var task model.Task
|
||
err = tx.GetContext(ctx, &task, `INSERT INTO tasks(id,kind,action,target_id,target_name,actor,status,message,idempotency_key) VALUES($1,$2,$3,$4,$5,$6,'queued','已提交，等待执行回执',$7) RETURNING *`, store.ID("task_"), kind, action, targetID, targetName, actor, key)
|
||
if err != nil {
|
||
if conflict(err) {
|
||
s.fail(c, 409, "task_conflict", "已有相同请求或该对象有任务执行中，请刷新任务列表")
|
||
} else {
|
||
s.dbError(c, err)
|
||
}
|
||
return
|
||
}
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO audit_logs VALUES($1,$2,$3,$4,'submitted',$5,now())", store.ID("audit_"), actor, action, targetName, c.GetString("requestId"))
|
||
if err == nil {
|
||
err = tx.Commit()
|
||
}
|
||
if err != nil {
|
||
s.dbError(c, err)
|
||
return
|
||
}
|
||
c.JSON(202, task)
|
||
}
|
||
|
||
// Complete persists the final result and its audit entry together.
|
||
func (s *Server) complete(ctx context.Context, t model.Task, status, message string) error {
|
||
tx, err := s.DB.BeginTxx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
if err = s.completeTx(ctx, tx, t, status, message); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
func (s *Server) completeTx(ctx context.Context, tx *sqlx.Tx, t model.Task, status, message string) error {
|
||
result, err := tx.ExecContext(ctx, "UPDATE tasks SET status=$1,message=$2,updated_at=now() WHERE id=$3 AND status IN ('running','queued')", status, message, t.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
n, _ := result.RowsAffected()
|
||
if n == 0 {
|
||
return nil
|
||
}
|
||
// Session release is scoped to a session ID by the device event endpoint.
|
||
// A delayed command acknowledgement must not clear a newer connection.
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO audit_logs VALUES($1,$2,$3,$4,$5,$6,now())", store.ID("audit_"), t.Actor, t.Action, t.TargetName, status, t.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
func ratio(a, b float64) float64 {
|
||
if b <= 0 {
|
||
return 0
|
||
}
|
||
return a / b
|
||
}
|
||
func (s *Server) sync(ctx context.Context) error {
|
||
if err := s.RefreshCompute(ctx); err != nil {
|
||
return err
|
||
}
|
||
pve := s.currentPVE()
|
||
if pve == nil {
|
||
return nil
|
||
}
|
||
items, err := pve.Resources(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
tx, err := s.DB.BeginTxx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
if _, err = tx.ExecContext(ctx, "DELETE FROM resources WHERE provider='proxmox'"); err != nil {
|
||
return err
|
||
}
|
||
for _, r := range items {
|
||
if r.Type == "node" {
|
||
if _, err = tx.ExecContext(ctx, `INSERT INTO resources VALUES($1,$1,$2,$3,$4,$5,'proxmox',now()) ON CONFLICT(id) DO UPDATE SET status=$2,cpu=$3,memory=$4,storage=$5,provider='proxmox',updated_at=now()`, r.Node, r.Status, r.CPU, ratio(r.Memory, r.MaxMemory), ratio(r.Disk, r.MaxDisk)); err != nil {
|
||
return err
|
||
}
|
||
if _, err = tx.ExecContext(ctx, `INSERT INTO resource_samples(resource_id,cpu,memory,storage) VALUES($1,CASE WHEN $7 THEN $2::double precision ELSE NULL END,CASE WHEN $7 AND $3::double precision>0 THEN $4::double precision ELSE NULL END,CASE WHEN $7 AND $5::double precision>0 THEN $6::double precision ELSE NULL END)`, r.Node, r.CPU, r.MaxMemory, ratio(r.Memory, r.MaxMemory), r.MaxDisk, ratio(r.Disk, r.MaxDisk), r.Status == "online"); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if r.Type == "qemu" {
|
||
if _, err = tx.ExecContext(ctx, "UPDATE desktops SET power_state=$1,updated_at=now() WHERE provider='proxmox' AND node=$2 AND vmid=$3", r.Status, r.Node, r.VMID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (s *Server) RunWorker(ctx context.Context) {
|
||
// Hold a session lock so only one worker can resume PVE tasks across API replicas.
|
||
conn, err := s.DB.Connx(ctx)
|
||
if err != nil {
|
||
return
|
||
}
|
||
defer conn.Close()
|
||
for {
|
||
var acquired bool
|
||
if err = conn.GetContext(ctx, &acquired, "SELECT pg_try_advisory_lock(7651092027)"); err != nil {
|
||
return
|
||
}
|
||
if acquired {
|
||
break
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(time.Second):
|
||
}
|
||
}
|
||
defer conn.ExecContext(context.Background(), "SELECT pg_advisory_unlock(7651092027)")
|
||
workerCtx, stop := context.WithCancel(ctx)
|
||
provisionDone := make(chan struct{}, 1)
|
||
provisionBusy := false
|
||
defer func() {
|
||
stop()
|
||
if provisionBusy {
|
||
<-provisionDone
|
||
}
|
||
}()
|
||
ticker := time.NewTicker(300 * time.Millisecond)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-provisionDone:
|
||
provisionBusy = false
|
||
case <-ticker.C:
|
||
if err = conn.PingContext(ctx); err != nil {
|
||
return
|
||
}
|
||
// Unknown outcomes are never automatically retried after a worker restart.
|
||
var expired []model.Task
|
||
if s.DB.SelectContext(ctx, &expired, "SELECT * FROM tasks WHERE status IN ('queued','running') AND ((kind NOT IN ('provision','hardware','snapshot','batch-desktop','gpu-prepare') AND created_at<now()-interval '2 minutes') OR (kind='gpu-prepare' AND created_at<now()-interval '5 minutes') OR (kind='batch-desktop' AND ((status='queued' AND created_at<now()-interval '2 hours') OR (status='running' AND updated_at<now()-interval '2 minutes'))) OR (kind IN ('provision','hardware','snapshot') AND updated_at<now()-interval '35 minutes'))") == nil {
|
||
for _, t := range expired {
|
||
if t.Kind == "gpu-prepare" {
|
||
var state gpuPrepareState
|
||
status := "timeout"
|
||
if json.Unmarshal([]byte(t.ProviderRef), &state) == nil && state.Phase == "queued" {
|
||
status = "failed"
|
||
}
|
||
_ = s.complete(ctx, t, status, "桌面准备超时，请核对任务和实际电源状态；不会自动重放")
|
||
continue
|
||
}
|
||
if t.Kind == "batch-desktop" {
|
||
s.finishBatchPower(t, "", "expired", "")
|
||
continue
|
||
}
|
||
if t.Kind == "snapshot" {
|
||
s.finishSnapshot(t, "", "expired", "")
|
||
continue
|
||
}
|
||
if t.Kind == "hardware" {
|
||
s.finishHardware(t, "", "expired", "")
|
||
continue
|
||
}
|
||
if t.Kind == "provision" {
|
||
_, _ = s.DB.ExecContext(ctx, "UPDATE provisioning_jobs SET state='uncertain' WHERE task_id=$1 AND state='reserved'", t.ID)
|
||
}
|
||
_ = s.complete(ctx, t, "timeout", "结果未确认，请核对设备或 PVE 实际状态；不会自动重放")
|
||
}
|
||
}
|
||
var t model.Task
|
||
err := s.DB.GetContext(ctx, &t, `UPDATE tasks SET status='running',updated_at=now(),message='正在执行，等待最终回执' WHERE id=(SELECT id FROM tasks WHERE ((status='queued' AND kind!='terminal') OR (kind='provision' AND status='running' AND provider_ref!='') OR (kind IN ('hardware','snapshot','batch-desktop') AND status='running')) AND ($1 OR kind NOT IN ('provision','hardware','snapshot','batch-desktop','gpu-prepare')) ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`, !provisionBusy)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
if t.Kind == "provision" || t.Kind == "hardware" || t.Kind == "snapshot" || t.Kind == "batch-desktop" || t.Kind == "gpu-prepare" {
|
||
provisionBusy = true
|
||
go func(task model.Task) { s.execute(workerCtx, task); provisionDone <- struct{}{} }(t)
|
||
} else {
|
||
s.execute(ctx, t)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Resource reads are independent of the command queue and never change VM power.
|
||
func (s *Server) RunSync(ctx context.Context) {
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
readCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
||
if err := s.sync(readCtx); err != nil {
|
||
log.Printf("PVE resource sync failed: %v", err)
|
||
}
|
||
cancel()
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
func (s *Server) execute(parent context.Context, t model.Task) {
|
||
if t.Kind == "gpu-prepare" {
|
||
s.executeGPUPrepare(parent, t)
|
||
return
|
||
}
|
||
if t.Kind == "batch-desktop" {
|
||
s.executeBatchPower(parent, t)
|
||
return
|
||
}
|
||
if t.Kind == "snapshot" {
|
||
s.executeSnapshot(parent, t)
|
||
return
|
||
}
|
||
if t.Kind == "hardware" {
|
||
s.executeHardware(parent, t)
|
||
return
|
||
}
|
||
if err := s.RefreshCompute(parent); err != nil {
|
||
_ = s.complete(parent, t, "failed", "计算平台配置无法加载，未提交操作")
|
||
return
|
||
}
|
||
pve := s.currentPVE()
|
||
if t.Kind == "provision" {
|
||
s.executeProvision(parent, t)
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(parent, 90*time.Second)
|
||
defer cancel()
|
||
var err error
|
||
uncertain := false
|
||
message := "已收到执行结果"
|
||
if t.Kind == "resource" {
|
||
err = s.sync(ctx)
|
||
} else {
|
||
var d model.Desktop
|
||
err = s.DB.GetContext(ctx, &d, "SELECT * FROM desktops WHERE id=$1", t.TargetID)
|
||
if err == nil && d.Provider == "demo" {
|
||
if s.Config.Mode != "demo" {
|
||
err = errors.New("demo disabled")
|
||
} else {
|
||
select {
|
||
case <-ctx.Done():
|
||
err = ctx.Err()
|
||
case <-time.After(500 * time.Millisecond):
|
||
}
|
||
if err == nil {
|
||
power := "running"
|
||
if t.Action == "shutdown" {
|
||
power = "stopped"
|
||
}
|
||
_, err = s.DB.ExecContext(ctx, "UPDATE desktops SET power_state=$1,service_state='unknown',updated_at=now() WHERE id=$2", power, d.ID)
|
||
message = "演示计算任务已完成；未检测真实连接服务"
|
||
}
|
||
}
|
||
} else if err == nil {
|
||
if pve == nil {
|
||
err = errors.New("PVE unconfigured")
|
||
} else {
|
||
var upid string
|
||
upid, err = pve.Action(ctx, d.Node, d.VMID, t.Action)
|
||
uncertain = true
|
||
var httpError *proxmox.HTTPError
|
||
if errors.As(err, &httpError) && httpError.Status >= 400 && httpError.Status < 500 {
|
||
uncertain = false
|
||
}
|
||
if err == nil {
|
||
_, err = s.DB.ExecContext(ctx, "UPDATE tasks SET provider_ref=$1 WHERE id=$2", upid, t.ID)
|
||
}
|
||
if err == nil {
|
||
err = pve.Wait(ctx, d.Node, upid, time.Second)
|
||
}
|
||
if err == nil {
|
||
var power string
|
||
power, err = pve.Power(ctx, d.Node, d.VMID)
|
||
if err == nil {
|
||
expected := "running"
|
||
if t.Action == "shutdown" {
|
||
expected = "stopped"
|
||
}
|
||
if power != expected {
|
||
err = fmt.Errorf("power result not confirmed")
|
||
} else {
|
||
_, err = s.DB.ExecContext(ctx, "UPDATE desktops SET power_state=$1,service_state='unknown',updated_at=now() WHERE id=$2", power, d.ID)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
status := "success"
|
||
if err != nil {
|
||
log.Printf("task execution error taskId=%s error=%v", t.ID, err)
|
||
status = "failed"
|
||
message = "执行失败，请检查服务日志及 PVE 任务；未自动重试"
|
||
if (uncertain && !errors.Is(err, proxmox.ErrTaskFailed)) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
|
||
status = "timeout"
|
||
message = "结果未确认，请核对实际状态；未自动重试"
|
||
}
|
||
}
|
||
finalCtx, finalCancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer finalCancel()
|
||
_ = s.complete(finalCtx, t, status, message)
|
||
}
|