xuans/Scert
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

补齐任务部署结果持久化与查询接口

Browse Source 
保存批量部署每台服务器的执行结果，新增数据库迁移、仓储查询和任务部署结果接口，并补充应用与仓储测试。
This commit is contained in:
xuans 2026-09-30 06:17:02 +08:00
parent 253f2bd824
commit 126cb6d566
11 changed files with 330 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
backend/internal/app/certificate_tasks.go
View File

@ -9,6 +9,7 @@ import (
"time"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
"git.pcux.cn/xuans/Scert/backend/internal/domain"
"git.pcux.cn/xuans/Scert/backend/internal/repository"
)
@ -148,6 +149,9 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
issueCtx, cancel := context.WithTimeout(taskCtx, 30*time.Minute)
result, resultErr = execute(issueCtx, id)
cancel()
if recordErr := w.tasks.RecordTaskDeploymentResults(taskCtx, task.ID, attempt.ID, number, taskDeploymentResults(result.Deployment)); recordErr != nil {
resultErr = errors.Join(resultErr, recordErr)
}
finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
finished := time.Now().UTC()
status, message := certificateTaskOutcome(result, resultErr)
@ -169,6 +173,24 @@ func (w *certificateTaskWorkflow) run(ctx context.Context, id, taskType, stage s
return result, resultErr
}
func taskDeploymentResults(report deployment.Report) []repository.TaskDeploymentResult {
if len(report.Results) == 0 {
return nil
}
items := make([]repository.TaskDeploymentResult, 0, len(report.Results))
for index, item := range report.Results {
items = append(items, repository.TaskDeploymentResult{
ResultIndex: index,
ServerID: item.ServerID,
Server: item.Server,
Status: item.Status,
Message: item.Message,
BackupID: item.BackupID,
})
}
return items
}
func certificateTaskOutcome(result CertificateWorkflowResult, err error) (string, string) {
if errors.Is(err, context.Canceled) {
return "cancelled", "\u8bc1\u4e66\u5de5\u4f5c\u6d41\u5df2\u53d6\u6d88"

31
backend/internal/app/certificate_tasks_test.go
View File

@ -421,3 +421,34 @@ func TestCertificateTasksRecordPartialDeploymentFailure(t *testing.T) {
}
assertCertificateTask(t, a, "partial_success", "partial_success")
}
func TestCertificateTaskPersistsDeploymentResults(t *testing.T) {
w, a := newCertificateTaskFixture(t)
w.workflow = &CertificateWorkflow{
Certificates: &workflowCertificateService{result: certificate.IssueResult{DomainID: "task-domain"}},
Domains: &workflowDomainState{},
Servers: workflowServerLister{servers: []deployment.Server{{ID: "server-1"}, {ID: "server-2"}}},
CurrentCertificate: func(context.Context, string) (certificate.IssueResult, error) {
return certificate.IssueResult{DomainID: "task-domain", Saved: certificate.SavedResource{CertificatePath: "cert.pem"}}, nil
},
Deployment: &workflowDeployer{report: deployment.Report{
Total: 2, Success: 1, Failed: 1, Status: "partial",
Results: []deployment.Result{
{ServerID: "server-1", Server: "第一台服务器", Status: "success", Message: "部署成功"},
{ServerID: "server-2", Server: "第二台服务器", Status: "failed", Message: "部署失败"},
},
}},
}
_, err := w.DeployCurrent(t.Context(), "task-domain")
if err == nil {
t.Fatal("部分部署失败应返回错误")
}
items, err := a.Tasks.ListTasks(t.Context(), "task-domain", 10)
if err != nil || len(items) != 1 {
t.Fatalf("任务读取失败：%+v %v", items, err)
}
results, err := a.Tasks.ListTaskDeploymentResults(t.Context(), items[0].ID)
if err != nil || len(results) != 2 || results[0].AttemptID == "" || results[1].Status != "failed" {
t.Fatalf("部署结果未持久化：%+v %v", results, err)
}
}

5
backend/internal/database/migrate.go
View File

@ -17,6 +17,7 @@ import (
//go:embed migrations/sqlite/0013_history_scope.sql migrations/mysql/0013_history_scope.sql migrations/postgres/0013_history_scope.sql
//go:embed migrations/sqlite/0014_project_members.sql migrations/mysql/0014_project_members.sql migrations/postgres/0014_project_members.sql
//go:embed migrations/sqlite/0015_task_idempotency.sql migrations/mysql/0015_task_idempotency.sql migrations/postgres/0015_task_idempotency.sql
//go:embed migrations/sqlite/0016_task_deployment_results.sql migrations/mysql/0016_task_deployment_results.sql migrations/postgres/0016_task_deployment_results.sql
var migrationFiles embed.FS
func Migrate(ctx context.Context, db *sql.DB, driver string) error {
@ -41,7 +42,7 @@ func Migrate(ctx context.Context, db *sql.DB, driver string) error {
return nil
}
const latestMigrationVersion = 15
const latestMigrationVersion = 16
// currentMigrationVersion 在执行业务迁移前拒绝新版数据库及不连续的迁移历史。
func currentMigrationVersion(ctx context.Context, db *sql.DB) (int, error) {
@ -223,6 +224,8 @@ func migrationName(version int) string {
return "project_members"
case 15:
return "task_idempotency"
case 16:
return "task_deployment_results"
default:
return "unknown"
}

14
backend/internal/database/migrations/mysql/0016_task_deployment_results.sql Normal file
View File

@ -0,0 +1,14 @@
CREATE TABLE task_deployment_results (
task_id VARCHAR(36) NOT NULL,
attempt_id VARCHAR(36) NOT NULL,
attempt_number INT NOT NULL,
result_index INT NOT NULL,
server_id VARCHAR(36) NOT NULL,
server_name VARCHAR(255) NOT NULL,
status VARCHAR(32) NOT NULL,
message TEXT NOT NULL,
backup_id VARCHAR(128) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (task_id, attempt_id, result_index),
INDEX idx_task_deployment_results_task_id (task_id)
);

15
backend/internal/database/migrations/postgres/0016_task_deployment_results.sql Normal file
View File

@ -0,0 +1,15 @@
CREATE TABLE task_deployment_results (
task_id VARCHAR(36) NOT NULL,
attempt_id VARCHAR(36) NOT NULL,
attempt_number INTEGER NOT NULL,
result_index INTEGER NOT NULL,
server_id VARCHAR(36) NOT NULL,
server_name VARCHAR(255) NOT NULL,
status VARCHAR(32) NOT NULL,
message TEXT NOT NULL,
backup_id VARCHAR(128) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (task_id, attempt_id, result_index)
);
CREATE INDEX idx_task_deployment_results_task_id ON task_deployment_results(task_id);

15
backend/internal/database/migrations/sqlite/0016_task_deployment_results.sql Normal file
View File

@ -0,0 +1,15 @@
CREATE TABLE task_deployment_results (
task_id TEXT NOT NULL,
attempt_id TEXT NOT NULL,
attempt_number INTEGER NOT NULL,
result_index INTEGER NOT NULL,
server_id TEXT NOT NULL,
server_name TEXT NOT NULL,
status TEXT NOT NULL,
message TEXT NOT NULL,
backup_id TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (task_id, attempt_id, result_index)
);
CREATE INDEX idx_task_deployment_results_task_id ON task_deployment_results(task_id);

1
backend/internal/httpapi/server.go
View File

@ -222,6 +222,7 @@ func (s *Server) registerRoutes() {
api.DELETE("/tasks/:id", s.cancelTask)
api.POST("/tasks/:id/retry", s.retryTask)
api.GET("/tasks/:id/attempts", s.listTaskAttempts)
api.GET("/tasks/:id/deployment-results", s.listTaskDeploymentResults)
adminAPI.GET("/settings", s.getSettings)
adminAPI.PUT("/settings", s.requireSuperAdminSystemConfig, s.updateSettings)

50
backend/internal/httpapi/task_api.go
View File

@ -13,6 +13,7 @@ import (
"github.com/gin-gonic/gin"
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
"git.pcux.cn/xuans/Scert/backend/internal/domain"
"git.pcux.cn/xuans/Scert/backend/internal/repository"
)
@ -177,6 +178,9 @@ func (s *Server) executeCertificateIssueTask(c *gin.Context, task repository.Tas
})
certificateResult, deploymentReport, issueErr := s.issueCertificate(executionCtx, task.DomainID)
deploymentReport = sanitizeDeploymentReport(deploymentReport)
if recordErr := s.tasks.RecordTaskDeploymentResults(context.Background(), task.ID, attempt.ID, attempt.AttemptNumber, taskDeploymentResults(deploymentReport)); recordErr != nil {
issueErr = errors.Join(issueErr, recordErr)
}
finished := time.Now().UTC()
status := "success"
message := "证书申请和部署完成"
@ -503,6 +507,52 @@ func (s *Server) listTaskAttempts(c *gin.Context) {
}
c.JSON(http.StatusOK, gin.H{"items": items})
}
func (s *Server) listTaskDeploymentResults(c *gin.Context) {
reader, ok := s.projectTaskReader(c)
if !ok {
return
}
actorID := c.GetString(authenticatedActorKey)
if _, err := reader.GetVisibleTask(c.Request.Context(), actorID, c.Param("id")); err != nil {
if errors.Is(err, sql.ErrNoRows) || errors.Is(err, repository.ErrProjectAccessDenied) {
s.writeError(c, http.StatusForbidden, "无权访问此任务", nil)
return
}
s.writeError(c, http.StatusInternalServerError, "读取任务失败", err)
return
}
items, err := reader.ListVisibleTaskDeploymentResults(c.Request.Context(), actorID, c.Param("id"))
if err != nil {
s.writeError(c, http.StatusInternalServerError, "查询任务部署结果失败", err)
return
}
for index := range items {
if items[index].Status == "failed" {
items[index].Message = "证书部署失败"
}
}
c.JSON(http.StatusOK, gin.H{"items": items})
}
func taskDeploymentResults(report deployment.Report) []repository.TaskDeploymentResult {
if len(report.Results) == 0 {
return nil
}
items := make([]repository.TaskDeploymentResult, 0, len(report.Results))
for index, item := range report.Results {
items = append(items, repository.TaskDeploymentResult{
ResultIndex: index,
ServerID: item.ServerID,
Server: item.Server,
Status: item.Status,
Message: item.Message,
BackupID: item.BackupID,
})
}
return items
}
func (s *Server) listAuditLogs(c *gin.Context) {
if !s.requireProjectResourceReader(c) {
return

33
backend/internal/repository/project_task_reader.go
View File

@ -16,6 +16,7 @@ type ProjectTaskReader interface {
ListVisibleTasks(context.Context, string, string, int) ([]Task, error)
GetVisibleTask(context.Context, string, string) (Task, error)
ListVisibleTaskAttempts(context.Context, string, string) ([]TaskAttempt, error)
ListVisibleTaskDeploymentResults(context.Context, string, string) ([]TaskDeploymentResult, error)
CancelVisibleTask(context.Context, string, string, string, *time.Time) error
RetryVisibleTask(context.Context, string, string, *time.Time) (Task, TaskAttempt, error)
}
@ -131,6 +132,38 @@ func (r *SQLTaskRepository) ListVisibleTaskAttempts(ctx context.Context, actorID
return items, nil
}
func (r *SQLTaskRepository) ListVisibleTaskDeploymentResults(ctx context.Context, actorID, taskID string) ([]TaskDeploymentResult, error) {
if r == nil || r.db == nil {
return nil, errors.New("数据库连接不能为空")
}
predicate, args := taskVisibilityPredicate("tasks", domain.PermissionTaskRead)
args[0] = strings.TrimSpace(actorID)
args = append(args, strings.TrimSpace(taskID))
rows, err := r.db.QueryContext(ctx, r.query(`SELECT results.task_id, results.attempt_id, results.attempt_number,
results.result_index, results.server_id, results.server_name, results.status, results.message,
results.backup_id, results.created_at
FROM task_deployment_results results JOIN tasks ON tasks.id = results.task_id
WHERE `+predicate+` AND results.task_id = ? ORDER BY results.attempt_number, results.result_index`), args...)
if err != nil {
return nil, fmt.Errorf("查询可见部署结果失败：%w", err)
}
defer rows.Close()
items := make([]TaskDeploymentResult, 0)
for rows.Next() {
var item TaskDeploymentResult
if err := rows.Scan(&item.TaskID, &item.AttemptID, &item.AttemptNumber, &item.ResultIndex,
&item.ServerID, &item.Server, &item.Status, &item.Message, &item.BackupID, &item.CreatedAt); err != nil {
return nil, fmt.Errorf("读取可见部署结果失败：%w", err)
}
item.CreatedAt = item.CreatedAt.UTC()
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("读取可见部署结果列表失败：%w", err)
}
return items, nil
}
func (r *SQLTaskRepository) CancelVisibleTask(ctx context.Context, actorID, taskID, message string, finishedAt *time.Time) error {
if r == nil || r.db == nil {
return errors.New("数据库连接不能为空")

108
backend/internal/repository/task_repository.go
View File

@ -41,6 +41,19 @@ type TaskAttempt struct {
CreatedAt time.Time `json:"created_at"`
}
type TaskDeploymentResult struct {
TaskID string `json:"task_id"`
AttemptID string `json:"attempt_id"`
AttemptNumber int `json:"attempt_number"`
ResultIndex int `json:"result_index"`
ServerID string `json:"server_id"`
Server string `json:"server"`
Status string `json:"status"`
Message string `json:"message"`
BackupID string `json:"backup_id,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
// ErrTaskAlreadyRunning 表示同一域名的同类任务仍处于活动状态。
var ErrTaskAlreadyRunning = errors.New("同一域名的同类任务正在执行")
@ -56,6 +69,8 @@ type TaskRepository interface {
CreateAttempt(context.Context, TaskAttempt) error
FinishAttempt(context.Context, string, string, string, *time.Time) error
ListAttempts(context.Context, string) ([]TaskAttempt, error)
RecordTaskDeploymentResults(context.Context, string, string, int, []TaskDeploymentResult) error
ListTaskDeploymentResults(context.Context, string) ([]TaskDeploymentResult, error)
}
type SQLTaskRepository struct {
@ -498,6 +513,99 @@ func (r *SQLTaskRepository) ListAttempts(ctx context.Context, taskID string) ([]
return result, nil
}
func (r *SQLTaskRepository) RecordTaskDeploymentResults(ctx context.Context, taskID, attemptID string, attemptNumber int, items []TaskDeploymentResult) error {
if r == nil || r.db == nil {
return errors.New("数据库连接不能为空")
}
taskID = strings.TrimSpace(taskID)
attemptID = strings.TrimSpace(attemptID)
if taskID == "" || attemptID == "" || attemptNumber <= 0 {
return errors.New("任务和任务尝试标识不能为空")
}
if len(items) == 0 {
return nil
}
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("开始保存部署结果事务失败：%w", err)
}
rollback := func() {
_ = tx.Rollback()
}
var attemptCount int
if err := tx.QueryRowContext(ctx, r.query(`SELECT COUNT(*) FROM task_attempts WHERE id = ? AND task_id = ? AND attempt_number = ?`), attemptID, taskID, attemptNumber).Scan(&attemptCount); err != nil {
rollback()
return fmt.Errorf("校验任务尝试失败：%w", err)
}
if attemptCount != 1 {
rollback()
return errors.New("任务尝试不存在或不匹配")
}
if _, err := tx.ExecContext(ctx, r.query(`DELETE FROM task_deployment_results WHERE task_id = ? AND attempt_id = ?`), taskID, attemptID); err != nil {
rollback()
return fmt.Errorf("清理旧部署结果失败：%w", err)
}
seen := make(map[int]struct{}, len(items))
for index, item := range items {
resultIndex := item.ResultIndex
if resultIndex < 0 {
resultIndex = index
}
if _, exists := seen[resultIndex]; exists {
rollback()
return errors.New("部署结果序号重复")
}
seen[resultIndex] = struct{}{}
createdAt := item.CreatedAt
if createdAt.IsZero() {
createdAt = time.Now().UTC()
}
if _, err := tx.ExecContext(ctx, r.query(`INSERT INTO task_deployment_results
(task_id, attempt_id, attempt_number, result_index, server_id, server_name, status, message, backup_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
taskID, attemptID, attemptNumber, resultIndex, strings.TrimSpace(item.ServerID), strings.TrimSpace(item.Server),
strings.TrimSpace(item.Status), item.Message, strings.TrimSpace(item.BackupID), createdAt.UTC()); err != nil {
rollback()
return fmt.Errorf("保存部署结果失败：%w", err)
}
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("提交部署结果事务失败：%w", err)
}
return nil
}
func (r *SQLTaskRepository) ListTaskDeploymentResults(ctx context.Context, taskID string) ([]TaskDeploymentResult, error) {
if r == nil || r.db == nil {
return nil, errors.New("数据库连接不能为空")
}
taskID = strings.TrimSpace(taskID)
if taskID == "" {
return nil, errors.New("任务 ID 不能为空")
}
rows, err := r.db.QueryContext(ctx, r.query(`SELECT task_id, attempt_id, attempt_number, result_index,
server_id, server_name, status, message, backup_id, created_at
FROM task_deployment_results WHERE task_id = ? ORDER BY attempt_number, result_index`), taskID)
if err != nil {
return nil, fmt.Errorf("查询部署结果失败：%w", err)
}
defer rows.Close()
items := make([]TaskDeploymentResult, 0)
for rows.Next() {
var item TaskDeploymentResult
if err := rows.Scan(&item.TaskID, &item.AttemptID, &item.AttemptNumber, &item.ResultIndex,
&item.ServerID, &item.Server, &item.Status, &item.Message, &item.BackupID, &item.CreatedAt); err != nil {
return nil, fmt.Errorf("读取部署结果失败：%w", err)
}
item.CreatedAt = item.CreatedAt.UTC()
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("读取部署结果列表失败：%w", err)
}
return items, nil
}
const taskSelect = `SELECT id, domain_id, task_type, status, stage, message, operator_id, certificate_version_id, idempotency_key, started_at, finished_at, created_at, updated_at, scope_type, project_id FROM tasks`
func scanTask(row rowScanner) (Task, error) {

37
backend/internal/repository/task_repository_test.go
View File

@ -89,6 +89,43 @@ func TestSQLTaskRepositoryLifecycle(t *testing.T) {
}
}
func TestSQLTaskRepositoryPersistsDeploymentResultsByAttempt(t *testing.T) {
repository, db := newTaskRepositoryTest(t)
defer db.Close()
ctx := context.Background()
if err := repository.CreateTask(ctx, Task{
ID: "task-deployment-results", DomainID: "domain-1", TaskType: "certificate_issue",
Status: "running", Stage: "certificate_issue", Message: "正在执行证书工作流",
}); err != nil {
t.Fatalf("创建部署任务失败：%v", err)
}
if err := repository.CreateAttempt(ctx, TaskAttempt{
ID: "attempt-deployment-results", TaskID: "task-deployment-results", AttemptNumber: 1,
Stage: "certificate_issue", Status: "running", Message: "正在执行证书工作流",
}); err != nil {
t.Fatalf("创建部署任务尝试失败：%v", err)
}
created := time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC)
if err := repository.RecordTaskDeploymentResults(ctx, "task-deployment-results", "attempt-deployment-results", 1, []TaskDeploymentResult{
{ResultIndex: 0, ServerID: "server-1", Server: "第一台服务器", Status: "success", Message: "部署成功", CreatedAt: created},
{ResultIndex: 1, ServerID: "server-2", Server: "第二台服务器", Status: "failed", Message: "部署失败", CreatedAt: created},
}); err != nil {
t.Fatalf("保存部署结果失败：%v", err)
}
items, err := repository.ListTaskDeploymentResults(ctx, "task-deployment-results")
if err != nil || len(items) != 2 || items[0].AttemptNumber != 1 || items[1].ServerID != "server-2" || items[1].Status != "failed" {
t.Fatalf("部署结果读取错误：%+v %v", items, err)
}
if err := repository.RecordTaskDeploymentResults(ctx, "task-deployment-results", "attempt-deployment-results", 1, []TaskDeploymentResult{
{ResultIndex: 0, ServerID: "server-1", Server: "第一台服务器", Status: "success", Message: "重新部署成功", CreatedAt: created},
}); err != nil {
t.Fatalf("替换部署结果失败：%v", err)
}
items, err = repository.ListTaskDeploymentResults(ctx, "task-deployment-results")
if err != nil || len(items) != 1 || items[0].Message != "重新部署成功" {
t.Fatalf("部署结果替换错误：%+v %v", items, err)
}
}
func TestSQLTaskRepositoryRejectsInvalidStatus(t *testing.T) {
repository, db := newTaskRepositoryTest(t)
defer db.Close()