diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 61e793d8..eeea583b 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -68,6 +68,11 @@ jobs: dataSourceName: root:casos-${{ github.run_id }}-${{ github.run_attempt }}@tcp(127.0.0.1:3306)/ dbName: casos socks5Proxy: "" + coreDNSImage: registry.k8s.io/coredns/coredns:v1.12.4 + flannelImage: ghcr.io/flannel-io/flannel:v0.27.4 + flannelCNIPluginImage: ghcr.io/flannel-io/flannel-cni-plugin:v1.8.0-flannel1 + localPathProvisionerImage: docker.io/rancher/local-path-provisioner:v0.0.32 + localPathHelperImage: docker.io/library/busybox:1.37.0 E2E_DATA_DIR: /tmp/casos-e2e-${{ github.run_id }}-${{ github.run_attempt }} E2E_APISERVER_PORT: 16443 E2E_WEBHOOK_PORT: 19443 diff --git a/controllers/helm.go b/controllers/helm.go index 7f19cbf3..88e8a92f 100644 --- a/controllers/helm.go +++ b/controllers/helm.go @@ -1,16 +1,24 @@ package controllers import ( + "context" + "crypto/sha256" "encoding/json" + "errors" "fmt" "io" "net/http" + "strconv" + "strings" "time" + "github.com/beego/beego/logs" "github.com/casosorg/casos/object" "github.com/casosorg/casos/store" ) +const helmOperationTaskNotFoundCode = "helm_task_not_found" + // ---------- ArtifactHub proxy ---------- type ahSearchResult struct { @@ -231,6 +239,13 @@ func (c *ApiController) InstallHelmChartStream() { c.StopRun() return } + owner := helmOperationOwner(c) + if owner == "" { + c.Ctx.ResponseWriter.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(c.Ctx.ResponseWriter.ResponseWriter, "data: ERROR: unable to identify Helm operation owner\n\n") + c.StopRun() + return + } w := c.Ctx.ResponseWriter.ResponseWriter w.Header().Set("Content-Type", "text/event-stream") @@ -238,20 +253,106 @@ func (c *ApiController) InstallHelmChartStream() { w.Header().Set("X-Accel-Buffering", "no") w.WriteHeader(http.StatusOK) - flusher, canFlush := w.(http.Flusher) ctx := c.Ctx.Request.Context() - logCh := store.InstallHelmChartStream(ctx, cfg, req.ReleaseName, req.Namespace, req.ChartName, req.RepoURL, req.Version, req.ValuesYAML) + task, err := object.CreateHelmOperationTask(owner, object.HelmOperationInstall, req.ReleaseName, req.Namespace, req.ChartName, req.Version) + if err != nil { + message := "unable to start Helm installation" + if errors.Is(err, object.ErrHelmOperationAlreadyActive) { + message = err.Error() + } else { + logs.Error("create Helm operation task: %v", err) + } + fmt.Fprintf(w, "data: ERROR: %s\n\n", message) + c.StopRun() + return + } + finishUnstartedTask := func(cause error) { + finishCtx, cancel := context.WithTimeout(context.Background(), object.HelmOperationPersistenceTimeout) + defer cancel() + if finishErr := object.FinishHelmOperationTaskContext(finishCtx, task.Id, false, cause.Error()); finishErr != nil { + logs.Error("finish unstarted Helm operation task %d: %v", task.Id, finishErr) + } + } + if _, err := fmt.Fprintf(w, "data: TASK_ID:%d\n\n", task.Id); err != nil { + finishUnstartedTask(fmt.Errorf("failed to send Helm operation task id: %w", err)) + c.StopRun() + return + } + responseController := http.NewResponseController(w) + if err := responseController.Flush(); err != nil { + finishUnstartedTask(fmt.Errorf("failed to flush Helm operation task id: %w", err)) + c.StopRun() + return + } + recorder := object.NewHelmOperationRecorder(task.Id) + logCh := store.InstallHelmChartStream(ctx, recorder, cfg, req.ReleaseName, req.Namespace, req.ChartName, req.RepoURL, req.Version, req.ValuesYAML) for line := range logCh { if _, err := fmt.Fprintf(w, "data: %s\n\n", line); err != nil { break } - if canFlush { - flusher.Flush() + if err := responseController.Flush(); err != nil { + break } } c.StopRun() } +// GetHelmOperationTask returns a persisted install task and its log history so +// an administrator can reconnect after an SSE stream is interrupted. +// @router /api/get-helm-operation-task [get] +func (c *ApiController) GetHelmOperationTask() { + if c.RequireAdmin() { + return + } + id, err := strconv.ParseInt(c.GetString("id"), 10, 64) + if err != nil || id <= 0 { + c.ResponseError("invalid task id") + return + } + owner := helmOperationOwner(c) + if owner == "" { + c.ResponseError("unable to identify Helm operation owner") + return + } + task, err := object.GetHelmOperationTaskForOwner(id, owner) + if err != nil { + logs.Error("get Helm operation task %d: %v", id, err) + c.ResponseError("failed to load Helm operation task") + return + } + if task == nil { + c.ResponseError("Helm operation task not found", helmOperationTaskNotFoundCode) + return + } + taskLogs, err := object.GetHelmOperationLogs(id, 1000) + if err != nil { + logs.Error("get Helm operation task %d logs: %v", id, err) + c.ResponseError("failed to load Helm operation task") + return + } + c.ResponseOk(task, taskLogs) +} + +func helmOperationOwner(c *ApiController) string { + if user := c.GetSessionUser(); user != nil { + return canonicalHelmOperationOwner(user.Id, user.Owner, user.Name) + } + return "" +} + +func canonicalHelmOperationOwner(id, owner, name string) string { + if id = strings.TrimSpace(id); id != "" { + return id + } + owner = strings.TrimSpace(owner) + name = strings.TrimSpace(name) + if owner == "" || name == "" { + return "" + } + digest := sha256.Sum256([]byte(owner + "\x00" + name)) + return fmt.Sprintf("casdoor:%x", digest) +} + // UpgradeHelmRelease upgrades an existing Helm release. // @router /api/upgrade-helm-release [post] func (c *ApiController) UpgradeHelmRelease() { diff --git a/deploy/installer.go b/deploy/installer.go index d5cdb9e7..48c59911 100644 --- a/deploy/installer.go +++ b/deploy/installer.go @@ -5,6 +5,8 @@ import ( "fmt" ) +const nodeDeployResolverPath = "/etc/casos-resolv.conf" + func (d *NodeDeployer) installNodeBinaries(ctx context.Context, runner *NodeDeploySSHRunner, arch, k8sVersion string) error { version := k8sVersion cniVersion := defaultNodeDeployCNIVersion @@ -28,6 +30,21 @@ sysctl --system >/dev/null test -e /proc/sys/net/bridge/bridge-nf-call-iptables`); err != nil { return fmt.Errorf("configure Kubernetes kernel networking: %w", err) } + if _, err := runner.RunRootContext(ctx, fmt.Sprintf(`set -e +if systemctl is-active --quiet systemd-resolved 2>/dev/null; then + for i in $(seq 1 30); do + [ -f /run/systemd/resolve/resolv.conf ] && break + sleep 1 + done + test -f /run/systemd/resolve/resolv.conf + resolver=/run/systemd/resolve/resolv.conf +else + resolver=/etc/resolv.conf +fi +ln -sfn "$resolver" %[1]s +test -f %[1]s`, nodeDeployResolverPath)); err != nil { + return fmt.Errorf("configure node resolver: %w", err) + } d.logStep(nodeDeployPhaseConfiguring, "Configuring containerd") if err := runner.WriteFileContext(ctx, "/etc/containerd/config.toml", GenerateContainerdConfig(d.config.SandboxImage, d.config.Socks5Proxy), "0644"); err != nil { diff --git a/deploy/preflight.go b/deploy/preflight.go index f18293a6..2888efdc 100644 --- a/deploy/preflight.go +++ b/deploy/preflight.go @@ -5,7 +5,9 @@ import ( "encoding/base64" "fmt" "net" + "net/http" "net/url" + "strconv" "strings" "time" ) @@ -90,16 +92,28 @@ func RunNodeDeployPreflight(ctx context.Context, runner *NodeDeploySSHRunner, ap // The bootstrap kubeconfig embeds the apiserver CA, but this early // reachability probe runs before those files exist on the target node. encodedURL := base64.StdEncoding.EncodeToString([]byte(trimmedURL)) - cmd := fmt.Sprintf("apiserver_url=$(printf %%s %s | base64 -d) && curl -kfsS --connect-timeout 5 \"$apiserver_url/readyz\" >/dev/null", shellSingleQuote(encodedURL)) - if _, err = runner.RunContext(ctx, cmd); err != nil { + cmd := fmt.Sprintf("apiserver_url=$(printf %%s %s | base64 -d) && curl -ksS --connect-timeout 5 --output /dev/null --write-out %%{http_code} \"$apiserver_url/readyz\"", shellSingleQuote(encodedURL)) + status, err := runner.RunContext(ctx, cmd) + if err != nil { return nil, fmt.Errorf("apiserver is not reachable from target: %w", err) } + if !isNodeDeployApiserverProbeStatus(status) { + return nil, fmt.Errorf("apiserver readiness probe returned HTTP status %q", strings.TrimSpace(status)) + } result.ApiserverOK = true } return result, nil } +func isNodeDeployApiserverProbeStatus(status string) bool { + code, err := strconv.Atoi(strings.TrimSpace(status)) + if err != nil { + return false + } + return (code >= 200 && code < 300) || code == http.StatusUnauthorized || code == http.StatusForbidden +} + func ResolveNodeDeployApiserverURL(ctx context.Context, runner *NodeDeploySSHRunner, fallbackURL string) string { fallbackURL = strings.TrimRight(strings.TrimSpace(fallbackURL), "/") if runner == nil { diff --git a/object/helm_operation.go b/object/helm_operation.go new file mode 100644 index 00000000..4959b9a9 --- /dev/null +++ b/object/helm_operation.go @@ -0,0 +1,332 @@ +package object + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "strings" + "time" + + "xorm.io/xorm" +) + +const ( + HelmOperationInstall = "install" + + HelmOperationStatusPending = "pending" + HelmOperationStatusRunning = "running" + HelmOperationStatusSucceeded = "succeeded" + HelmOperationStatusFailed = "failed" + + HelmOperationPhaseQueued = "queued" + HelmOperationPhaseLoading = "loading" + HelmOperationPhaseInstalling = "installing" + HelmOperationPhaseReady = "ready" + HelmOperationPhaseFailed = "failed" + + HelmOperationLogLevelInfo = "info" + HelmOperationLogLevelError = "error" + + HelmOperationPersistenceTimeout = 5 * time.Second + helmOperationStaleAfter = 11 * time.Minute +) + +var ( + ErrHelmOperationAlreadyActive = errors.New("Helm operation already active") + ErrHelmOperationAlreadyFinished = errors.New("Helm operation already finished") +) + +type HelmOperationTask struct { + Id int64 `xorm:"pk autoincr" json:"id"` + ActiveKey *string `xorm:"char(64) unique" json:"-"` + Owner string `xorm:"varchar(100) notnull index" json:"owner"` + Operation string `xorm:"varchar(30) notnull" json:"operation"` + ReleaseName string `xorm:"varchar(253) notnull index" json:"releaseName"` + Namespace string `xorm:"varchar(253) notnull index" json:"namespace"` + ChartName string `xorm:"varchar(253) notnull" json:"chartName"` + Version string `xorm:"varchar(100)" json:"version"` + Status string `xorm:"varchar(30) notnull index" json:"status"` + Phase string `xorm:"varchar(30) notnull" json:"phase"` + ErrorMsg string `xorm:"text" json:"errorMsg"` + CreatedAt time.Time `json:"createdAt"` + StartedAt time.Time `json:"startedAt"` + FinishedAt time.Time `json:"finishedAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type HelmOperationLog struct { + Id int64 `xorm:"pk autoincr" json:"id"` + TaskId int64 `xorm:"notnull index" json:"taskId"` + Level string `xorm:"varchar(20) notnull" json:"level"` + Message string `xorm:"text" json:"message"` + CreatedAt time.Time `json:"createdAt"` +} + +func CreateHelmOperationTask(owner, operation, releaseName, namespace, chartName, version string) (*HelmOperationTask, error) { + owner = strings.TrimSpace(owner) + operation = strings.TrimSpace(operation) + releaseName = strings.TrimSpace(releaseName) + namespace = strings.TrimSpace(namespace) + chartName = strings.TrimSpace(chartName) + if owner == "" || operation == "" || releaseName == "" || namespace == "" || chartName == "" { + return nil, fmt.Errorf("owner, operation, releaseName, namespace, and chartName are required") + } + if operation != HelmOperationInstall { + return nil, fmt.Errorf("unsupported Helm operation: %s", operation) + } + activeKey := helmOperationActiveKey(namespace, releaseName) + + result, err := withHelmOperationTransaction(func(session *xorm.Session) (interface{}, error) { + active := &HelmOperationTask{} + found, err := session. + Where("namespace = ? AND release_name = ? AND status IN (?, ?)", namespace, releaseName, HelmOperationStatusPending, HelmOperationStatusRunning). + Desc("id"). + ForUpdate(). + Get(active) + if err != nil { + return nil, err + } + if found { + if active.UpdatedAt.After(time.Now().UTC().Add(-helmOperationStaleAfter)) { + return nil, fmt.Errorf("%w: task %d is already active for %s/%s", ErrHelmOperationAlreadyActive, active.Id, namespace, releaseName) + } + now := time.Now().UTC() + if _, err := session.ID(active.Id). + Cols("active_key", "status", "phase", "error_msg", "finished_at", "updated_at"). + Update(&HelmOperationTask{ + ActiveKey: nil, + Status: HelmOperationStatusFailed, + Phase: HelmOperationPhaseFailed, + ErrorMsg: "Helm operation expired before completion", + FinishedAt: now, + UpdatedAt: now, + }); err != nil { + return nil, err + } + } + + now := time.Now().UTC() + task := &HelmOperationTask{ + ActiveKey: &activeKey, + Owner: owner, + Operation: operation, + ReleaseName: releaseName, + Namespace: namespace, + ChartName: chartName, + Version: strings.TrimSpace(version), + Status: HelmOperationStatusPending, + Phase: HelmOperationPhaseQueued, + CreatedAt: now, + UpdatedAt: now, + } + if _, err := session.Insert(task); err != nil { + conflict := &HelmOperationTask{} + if found, lookupErr := session.Where("active_key = ?", activeKey).Get(conflict); lookupErr == nil && found { + return nil, fmt.Errorf("%w: task %d is already active for %s/%s", ErrHelmOperationAlreadyActive, conflict.Id, namespace, releaseName) + } + return nil, err + } + return task, nil + }) + if err != nil { + return nil, err + } + task, ok := result.(*HelmOperationTask) + if !ok || task == nil { + return nil, fmt.Errorf("create Helm operation task returned invalid result") + } + return task, nil +} + +func helmOperationActiveKey(namespace, releaseName string) string { + return fmt.Sprintf("%x", sha256.Sum256([]byte(namespace+"\x00"+releaseName))) +} + +func GetHelmOperationTask(id int64) (*HelmOperationTask, error) { + task := &HelmOperationTask{Id: id} + found, err := ormer.Engine.Get(task) + if err != nil { + return nil, err + } + if !found { + return nil, nil + } + return task, nil +} + +// GetHelmOperationTaskForOwner deliberately returns (nil, nil) for both a +// missing task and an owner mismatch so callers cannot reveal another owner's +// task through their response behavior. +func GetHelmOperationTaskForOwner(id int64, owner string) (*HelmOperationTask, error) { + owner = strings.TrimSpace(owner) + if owner == "" { + return nil, fmt.Errorf("Helm operation owner is required") + } + task, err := GetHelmOperationTask(id) + if err != nil || task == nil { + return task, err + } + if task.Owner != owner { + return nil, nil + } + return task, nil +} + +func StartHelmOperationTaskContext(ctx context.Context, id int64, phase string) error { + if phase != HelmOperationPhaseLoading { + return fmt.Errorf("invalid Helm operation start phase: %s", phase) + } + now := time.Now().UTC() + affected, err := ormer.Engine.Context(ctx).ID(id). + Where("status = ? AND phase = ?", HelmOperationStatusPending, HelmOperationPhaseQueued). + Cols("status", "phase", "started_at", "updated_at"). + Update(&HelmOperationTask{Status: HelmOperationStatusRunning, Phase: phase, StartedAt: now, UpdatedAt: now}) + if err != nil { + return err + } + if affected == 0 { + return fmt.Errorf("Helm operation task %d is not pending", id) + } + return nil +} + +func UpdateHelmOperationTaskPhase(id int64, phase string) error { + return UpdateHelmOperationTaskPhaseContext(context.Background(), id, phase) +} + +func UpdateHelmOperationTaskPhaseContext(ctx context.Context, id int64, phase string) error { + if phase != HelmOperationPhaseInstalling { + return fmt.Errorf("invalid Helm operation phase: %s", phase) + } + affected, err := ormer.Engine.Context(ctx).ID(id). + Where("status = ? AND phase = ?", HelmOperationStatusRunning, HelmOperationPhaseLoading). + Cols("phase", "updated_at"). + Update(&HelmOperationTask{Phase: phase, UpdatedAt: time.Now().UTC()}) + if err != nil { + return err + } + if affected == 0 { + return fmt.Errorf("Helm operation task %d is not running", id) + } + return nil +} + +func FinishHelmOperationTask(id int64, success bool, errorMsg string) error { + return FinishHelmOperationTaskContext(context.Background(), id, success, errorMsg) +} + +func FinishHelmOperationTaskContext(ctx context.Context, id int64, success bool, errorMsg string) error { + status := HelmOperationStatusSucceeded + phase := HelmOperationPhaseReady + if success { + errorMsg = "" + } + if !success { + status = HelmOperationStatusFailed + phase = HelmOperationPhaseFailed + } + now := time.Now().UTC() + affected, err := ormer.Engine.Context(ctx).ID(id). + Where("status IN (?, ?)", HelmOperationStatusPending, HelmOperationStatusRunning). + Cols("active_key", "status", "phase", "error_msg", "finished_at", "updated_at"). + Update(&HelmOperationTask{ActiveKey: nil, Status: status, Phase: phase, ErrorMsg: errorMsg, FinishedAt: now, UpdatedAt: now}) + if err != nil { + return err + } + if affected == 0 { + return fmt.Errorf("%w: task %d", ErrHelmOperationAlreadyFinished, id) + } + return nil +} + +func HelmOperationTaskHasTerminalOutcomeContext(ctx context.Context, id int64, success bool, errorMsg string) (bool, error) { + task := &HelmOperationTask{Id: id} + found, err := ormer.Engine.Context(ctx).Get(task) + if err != nil || !found { + return false, err + } + if success { + return task.Status == HelmOperationStatusSucceeded && task.Phase == HelmOperationPhaseReady && task.ErrorMsg == "", nil + } + return task.Status == HelmOperationStatusFailed && task.Phase == HelmOperationPhaseFailed && task.ErrorMsg == errorMsg, nil +} + +func addHelmOperationLogs(taskID int64, logs []*HelmOperationLog) error { + return addHelmOperationLogsContext(context.Background(), taskID, logs) +} + +func addHelmOperationLogsContext(ctx context.Context, taskID int64, logs []*HelmOperationLog) error { + if len(logs) == 0 { + return nil + } + now := time.Now().UTC() + for _, entry := range logs { + if entry == nil || entry.TaskId != taskID { + return fmt.Errorf("Helm operation log task id does not match task %d", taskID) + } + if entry.CreatedAt.IsZero() { + entry.CreatedAt = now + } + } + _, err := withHelmOperationTransactionContext(ctx, func(session *xorm.Session) (interface{}, error) { + if _, err := session.Insert(&logs); err != nil { + return nil, err + } + _, err := session.ID(taskID). + Where("status IN (?, ?)", HelmOperationStatusPending, HelmOperationStatusRunning). + Cols("updated_at"). + Update(&HelmOperationTask{UpdatedAt: now}) + return nil, err + }) + return err +} + +// GetHelmOperationLogs returns the most recent limit entries in chronological +// order (oldest to newest within that window). +func GetHelmOperationLogs(taskID int64, limit int) ([]*HelmOperationLog, error) { + if taskID <= 0 { + return nil, fmt.Errorf("invalid taskId") + } + if limit <= 0 { + limit = 500 + } + if limit > 1000 { + return nil, fmt.Errorf("limit must not exceed 1000") + } + logs := []*HelmOperationLog{} + err := ormer.Engine.Where("task_id = ?", taskID).Desc("id").Limit(limit).Find(&logs) + for left, right := 0, len(logs)-1; left < right; left, right = left+1, right-1 { + logs[left], logs[right] = logs[right], logs[left] + } + return logs, err +} + +func withHelmOperationTransaction(fn func(*xorm.Session) (interface{}, error)) (interface{}, error) { + return withHelmOperationTransactionContext(context.Background(), fn) +} + +func withHelmOperationTransactionContext(ctx context.Context, fn func(*xorm.Session) (interface{}, error)) (interface{}, error) { + session := ormer.Engine.NewSession().Context(ctx) + defer func() { + if v := recover(); v != nil { + _ = session.Rollback() + session.Close() + panic(v) + } + session.Close() + }() + if err := session.Begin(); err != nil { + return nil, err + } + result, err := fn(session) + if err != nil { + _ = session.Rollback() + return nil, err + } + if err := session.Commit(); err != nil { + _ = session.Rollback() + return nil, err + } + return result, nil +} diff --git a/object/helm_operation_recorder.go b/object/helm_operation_recorder.go new file mode 100644 index 00000000..7b7d97d7 --- /dev/null +++ b/object/helm_operation_recorder.go @@ -0,0 +1,252 @@ +package object + +import ( + "context" + "errors" + "fmt" + "runtime/debug" + "strings" + "sync" + "time" + + "github.com/beego/beego/logs" +) + +const ( + helmOperationLogBatchSize = 50 + helmOperationLogFlushInterval = 200 * time.Millisecond + helmOperationLogEnqueueTimeout = 2 * time.Second + helmOperationLogPersistTimeout = HelmOperationPersistenceTimeout + helmOperationRecorderShutdown = 3*helmOperationLogPersistTimeout + time.Second + helmOperationRecorderStopGrace = 100 * time.Millisecond + helmOperationFinishTimeout = HelmOperationPersistenceTimeout + helmOperationFinishAttempts = 3 + helmOperationFinishRetryDelay = 100 * time.Millisecond +) + +type HelmOperationRecorder struct { + taskID int64 + queue chan *HelmOperationLog + done chan struct{} + + mu sync.Mutex + closed bool + batchErr error + writers sync.WaitGroup + finish sync.Once + finishErr error + + persistLogs func(context.Context, int64, []*HelmOperationLog) error + finishTask func(context.Context, int64, bool, string) error + terminalMatches func(context.Context, int64, bool, string) (bool, error) + persistTimeout time.Duration + shutdownTimeout time.Duration + finishTimeout time.Duration + finishRetryDelay time.Duration + persistCtx context.Context + cancelPersist context.CancelFunc +} + +func NewHelmOperationRecorder(taskID int64) *HelmOperationRecorder { + persistCtx, cancelPersist := context.WithCancel(context.Background()) + recorder := &HelmOperationRecorder{ + taskID: taskID, + queue: make(chan *HelmOperationLog, helmOperationLogBatchSize*2), + done: make(chan struct{}), + persistLogs: addHelmOperationLogsContext, + finishTask: FinishHelmOperationTaskContext, + terminalMatches: HelmOperationTaskHasTerminalOutcomeContext, + persistTimeout: helmOperationLogPersistTimeout, + shutdownTimeout: helmOperationRecorderShutdown, + finishTimeout: helmOperationFinishTimeout, + finishRetryDelay: helmOperationFinishRetryDelay, + persistCtx: persistCtx, + cancelPersist: cancelPersist, + } + go recorder.run() + return recorder +} + +func (r *HelmOperationRecorder) StartLoading() error { + ctx, cancel := context.WithTimeout(context.Background(), r.persistTimeout) + defer cancel() + return StartHelmOperationTaskContext(ctx, r.taskID, HelmOperationPhaseLoading) +} + +func (r *HelmOperationRecorder) MarkInstalling() error { + ctx, cancel := context.WithTimeout(context.Background(), r.persistTimeout) + defer cancel() + return UpdateHelmOperationTaskPhaseContext(ctx, r.taskID, HelmOperationPhaseInstalling) +} + +func (r *HelmOperationRecorder) RecordLog(line string) error { + level := HelmOperationLogLevelInfo + trimmed := strings.TrimSpace(line) + if len(trimmed) >= len("ERROR:") && strings.EqualFold(trimmed[:len("ERROR:")], "ERROR:") { + level = HelmOperationLogLevelError + } + entry := &HelmOperationLog{ + TaskId: r.taskID, + Level: level, + Message: line, + CreatedAt: time.Now().UTC(), + } + + r.mu.Lock() + if r.closed { + r.mu.Unlock() + return fmt.Errorf("Helm operation recorder is closed") + } + r.writers.Add(1) + r.mu.Unlock() + defer r.writers.Done() + timer := time.NewTimer(helmOperationLogEnqueueTimeout) + defer timer.Stop() + select { + case r.queue <- entry: + return nil + case <-timer.C: + return fmt.Errorf("Helm operation log queue is full") + } +} + +func (r *HelmOperationRecorder) Finish(installErr error) error { + r.finish.Do(func() { + r.mu.Lock() + r.closed = true + r.mu.Unlock() + + r.writers.Wait() + close(r.queue) + shutdownTimer := time.NewTimer(r.shutdownTimeout) + select { + case <-r.done: + shutdownTimer.Stop() + r.cancelPersist() + case <-shutdownTimer.C: + shutdownTimer.Stop() + r.setBatchErr(fmt.Errorf("timed out waiting for Helm operation log persistence")) + r.cancelPersist() + stopTimer := time.NewTimer(helmOperationRecorderStopGrace) + select { + case <-r.done: + stopTimer.Stop() + case <-stopTimer.C: + stopTimer.Stop() + // A database driver that ignores context cancellation may leave its + // call running; task completion remains bounded and the timeout is logged. + logs.Warning("Helm operation task %d log persistence did not stop after cancellation", r.taskID) + } + } + + r.mu.Lock() + batchErr := r.batchErr + r.mu.Unlock() + // Log durability is observable but does not change the outcome of a Helm + // install that already completed in the cluster. + success := installErr == nil + errorMsg := "" + if installErr != nil { + errorMsg = installErr.Error() + } + finishErr := r.finishTaskWithRetry(success, errorMsg) + if batchErr != nil { + logs.Warning("persist Helm operation task %d logs: %v", r.taskID, batchErr) + if finishErr != nil { + r.finishErr = fmt.Errorf("persist Helm operation logs: %v; finish task: %w", batchErr, finishErr) + return + } + } + r.finishErr = finishErr + }) + return r.finishErr +} + +func (r *HelmOperationRecorder) finishTaskWithRetry(success bool, errorMsg string) error { + // The install intentionally outlives the browser request, so terminal-state + // persistence uses its own bounded deadline instead of the request context. + var err error + for attempt := 0; attempt < helmOperationFinishAttempts; attempt++ { + ctx, cancel := context.WithTimeout(context.Background(), r.finishTimeout) + err = r.finishTask(ctx, r.taskID, success, errorMsg) + if err == nil { + cancel() + return nil + } + if errors.Is(err, ErrHelmOperationAlreadyFinished) { + cancel() + matchCtx, matchCancel := context.WithTimeout(context.Background(), r.finishTimeout) + matches, matchErr := r.terminalMatches(matchCtx, r.taskID, success, errorMsg) + matchCancel() + if matchErr != nil { + return fmt.Errorf("verify existing Helm operation terminal state: %w", matchErr) + } + if matches { + return nil + } + return err + } + cancel() + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return err + } + if attempt+1 < helmOperationFinishAttempts { + time.Sleep(r.finishRetryDelay) + } + } + return err +} + +func (r *HelmOperationRecorder) setBatchErr(err error) { + if err == nil { + return + } + r.mu.Lock() + if r.batchErr == nil { + r.batchErr = err + } + r.mu.Unlock() +} + +func (r *HelmOperationRecorder) run() { + defer func() { + if recovered := recover(); recovered != nil { + stack := debug.Stack() + logs.Error("persist Helm operation task %d logs panic: %v\n%s", r.taskID, recovered, stack) + r.setBatchErr(fmt.Errorf("persist Helm operation logs panic: %v", recovered)) + } + close(r.done) + }() + ticker := time.NewTicker(helmOperationLogFlushInterval) + defer ticker.Stop() + batch := make([]*HelmOperationLog, 0, helmOperationLogBatchSize) + flush := func() { + if len(batch) == 0 { + return + } + logs := append([]*HelmOperationLog(nil), batch...) + ctx, cancel := context.WithTimeout(r.persistCtx, r.persistTimeout) + err := r.persistLogs(ctx, r.taskID, logs) + cancel() + if err != nil { + r.setBatchErr(err) + } + batch = make([]*HelmOperationLog, 0, helmOperationLogBatchSize) + } + + for { + select { + case entry, ok := <-r.queue: + if !ok { + flush() + return + } + batch = append(batch, entry) + if len(batch) >= helmOperationLogBatchSize { + flush() + } + case <-ticker.C: + flush() + } + } +} diff --git a/object/ormer.go b/object/ormer.go index 414598c5..a891ad52 100644 --- a/object/ormer.go +++ b/object/ormer.go @@ -174,12 +174,18 @@ func (a *Ormer) close() { func (a *Ormer) createTable() { showSql := conf.GetConfigBool("showSql") a.Engine.ShowSQL(showSql) - _ = a.Engine.Sync2(new(Site)) - _ = a.Engine.Sync2(new(Machine)) - _ = a.Engine.Sync2(new(MachineNodeDeployTask)) - _ = a.Engine.Sync2(new(MachineNodeDeployLog)) - _ = a.Engine.Sync2(new(MachineNodeDeployCredential)) - _ = a.Engine.Sync2(new(CasbinRule)) - _ = a.Engine.Sync2(new(TrivyScanResult)) - _ = a.Engine.Sync2(new(HelmRepo)) + if err := a.Engine.Sync2( + new(Site), + new(Machine), + new(MachineNodeDeployTask), + new(MachineNodeDeployLog), + new(MachineNodeDeployCredential), + new(HelmOperationTask), + new(HelmOperationLog), + new(CasbinRule), + new(TrivyScanResult), + new(HelmRepo), + ); err != nil { + panic(fmt.Errorf("sync database schema: %w", err)) + } } diff --git a/routers/router.go b/routers/router.go index 9f46df91..82287d46 100644 --- a/routers/router.go +++ b/routers/router.go @@ -150,6 +150,7 @@ func InitAPI() { beego.Router("/api/get-helm-releases", &controllers.ApiController{}, "GET:GetHelmReleases") beego.Router("/api/install-helm-chart", &controllers.ApiController{}, "POST:InstallHelmChart") beego.Router("/api/install-helm-chart-stream", &controllers.ApiController{}, "POST:InstallHelmChartStream") + beego.Router("/api/get-helm-operation-task", &controllers.ApiController{}, "GET:GetHelmOperationTask") beego.Router("/api/upgrade-helm-release", &controllers.ApiController{}, "POST:UpgradeHelmRelease") beego.Router("/api/rollback-helm-release", &controllers.ApiController{}, "POST:RollbackHelmRelease") beego.Router("/api/uninstall-helm-release", &controllers.ApiController{}, "POST:UninstallHelmRelease") diff --git a/server/apiserver.go b/server/apiserver.go index 3a20c425..0dbd0c8a 100644 --- a/server/apiserver.go +++ b/server/apiserver.go @@ -22,6 +22,11 @@ import ( "k8s.io/kubernetes/cmd/kube-apiserver/app/options" ) +const ( + serviceClusterIPRange = "10.43.0.0/16" + kubernetesServiceIP = "10.43.0.1" +) + // Start launches kine and the apiserver in-process. // The returned channel is closed once the apiserver /readyz endpoint responds 200. func Start(ctx context.Context, cfg Config) (<-chan struct{}, error) { @@ -142,7 +147,7 @@ func buildApiserverArgs(cfg Config, certDir, etcdEndpoint, authzKubeconfig strin "--bind-address=0.0.0.0", fmt.Sprintf("--secure-port=%d", cfg.ApiserverPort), "--etcd-servers=" + etcdEndpoint, - "--service-cluster-ip-range=10.43.0.0/16", + "--service-cluster-ip-range=" + serviceClusterIPRange, "--allow-privileged=true", "--authorization-mode=" + authzMode(authzKubeconfig), "--enable-admission-plugins=NodeRestriction,ValidatingAdmissionWebhook", diff --git a/server/certs.go b/server/certs.go index 86158e8b..d933a902 100644 --- a/server/certs.go +++ b/server/certs.go @@ -496,7 +496,7 @@ func uniqueIPs(addrs ...string) []net.IP { } func desiredAPIServerIPs(ip, advertiseIP string) []net.IP { - return uniqueIPs(append([]string{"127.0.0.1", ip, advertiseIP}, allInterfaceIPs()...)...) + return uniqueIPs(append([]string{"127.0.0.1", kubernetesServiceIP, ip, advertiseIP}, allInterfaceIPs()...)...) } func apiServerServingCertUsable(certFile, keyFile string, desiredIPs []net.IP, caCert *x509.Certificate) bool { diff --git a/server/dns_bootstrap.go b/server/dns_bootstrap.go index ff02facb..940ba9fa 100644 --- a/server/dns_bootstrap.go +++ b/server/dns_bootstrap.go @@ -3,21 +3,26 @@ package server import ( "context" "fmt" + "strconv" + "time" "github.com/sirupsen/logrus" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" ) const ( clusterDNSNamespace = "kube-system" clusterDNSName = "coredns" + clusterDNSServiceIP = "10.43.0.10" // coreDNSRolloutRev forces a rollout when the managed CoreDNS pod template changes. coreDNSRolloutRev = "4" ) @@ -144,7 +149,8 @@ func ensureCoreDNSService(ctx context.Context, client kubernetes.Interface) erro Labels: coreDNSLabels(), }, Spec: corev1.ServiceSpec{ - Selector: coreDNSSelectorLabels(), + ClusterIP: clusterDNSServiceIP, + Selector: coreDNSSelectorLabels(), Ports: []corev1.ServicePort{ {Name: "dns", Port: 53, Protocol: corev1.ProtocolUDP, TargetPort: intstr.FromInt(53)}, {Name: "dns-tcp", Port: 53, Protocol: corev1.ProtocolTCP, TargetPort: intstr.FromInt(53)}, @@ -152,9 +158,30 @@ func ensureCoreDNSService(ctx context.Context, client kubernetes.Interface) erro }, }, } + current, err := client.CoreV1().Services(clusterDNSNamespace).Get(ctx, svc.Name, metav1.GetOptions{}) + if err == nil && current.Spec.ClusterIP != "" && current.Spec.ClusterIP != clusterDNSServiceIP { + if err := client.CoreV1().Services(clusterDNSNamespace).Delete(ctx, svc.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("replace CoreDNS service %s/%s: %w", clusterDNSNamespace, svc.Name, err) + } + if err := waitForServiceDeleted(ctx, client, clusterDNSNamespace, svc.Name); err != nil { + return fmt.Errorf("wait to replace CoreDNS service %s/%s: %w", clusterDNSNamespace, svc.Name, err) + } + } else if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("get CoreDNS service %s/%s: %w", clusterDNSNamespace, svc.Name, err) + } return createOrUpdateService(ctx, client, svc) } +func waitForServiceDeleted(ctx context.Context, client kubernetes.Interface, namespace, name string) error { + return wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 30*time.Second, true, func(ctx context.Context) (bool, error) { + _, err := client.CoreV1().Services(namespace).Get(ctx, name, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return true, nil + } + return false, err + }) +} + func ensureCoreDNSDeployment(ctx context.Context, client kubernetes.Interface, cfg Config) error { replicas := int32(1) maxSurge := intstr.FromInt(1) @@ -191,6 +218,7 @@ func ensureCoreDNSDeployment(ctx context.Context, client kubernetes.Interface, c {Key: "CriticalAddonsOnly", Operator: corev1.TolerationOpExists}, {Key: "node-role.kubernetes.io/control-plane", Operator: corev1.TolerationOpExists, Effect: corev1.TaintEffectNoSchedule}, {Key: "node-role.kubernetes.io/master", Operator: corev1.TolerationOpExists, Effect: corev1.TaintEffectNoSchedule}, + {Key: "casos.io/bootstrap", Operator: corev1.TolerationOpExists, Effect: corev1.TaintEffectNoSchedule}, }, Containers: []corev1.Container{ { @@ -198,6 +226,7 @@ func ensureCoreDNSDeployment(ctx context.Context, client kubernetes.Interface, c Image: cfg.CoreDNSImage, ImagePullPolicy: corev1.PullIfNotPresent, Args: []string{"-conf", "/etc/coredns/Corefile"}, + Env: coreDNSEnv(cfg), Ports: []corev1.ContainerPort{ {Name: "dns", ContainerPort: 53, Protocol: corev1.ProtocolUDP}, {Name: "dns-tcp", ContainerPort: 53, Protocol: corev1.ProtocolTCP}, @@ -283,6 +312,17 @@ func ensureCoreDNSDeployment(ctx context.Context, client kubernetes.Interface, c return createOrUpdateDeployment(ctx, client, deployment) } +func coreDNSEnv(cfg Config) []corev1.EnvVar { + if cfg.AdvertiseAddress == "" || cfg.ApiserverPort <= 0 { + return nil + } + return []corev1.EnvVar{ + {Name: "KUBERNETES_SERVICE_HOST", Value: cfg.AdvertiseAddress}, + {Name: "KUBERNETES_SERVICE_PORT", Value: strconv.Itoa(cfg.ApiserverPort)}, + {Name: "KUBERNETES_SERVICE_PORT_HTTPS", Value: strconv.Itoa(cfg.ApiserverPort)}, + } +} + func coreDNSLabels() map[string]string { labels := coreDNSSelectorLabels() labels["k8s-app"] = "kube-dns" @@ -357,6 +397,11 @@ func createOrUpdateService(ctx context.Context, client kubernetes.Interface, svc svc.Spec.LoadBalancerClass = current.Spec.LoadBalancerClass svc.Spec.LoadBalancerSourceRanges = current.Spec.LoadBalancerSourceRanges svc.Spec.AllocateLoadBalancerNodePorts = current.Spec.AllocateLoadBalancerNodePorts + if apiequality.Semantic.DeepEqual(current.Labels, svc.Labels) && + apiequality.Semantic.DeepEqual(current.Annotations, svc.Annotations) && + apiequality.Semantic.DeepEqual(current.Spec, svc.Spec) { + return nil + } if _, err := client.CoreV1().Services(svc.Namespace).Update(ctx, svc, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("update service %s/%s: %w", svc.Namespace, svc.Name, err) } diff --git a/store/helm.go b/store/helm.go index 09f90097..88c5d1d5 100644 --- a/store/helm.go +++ b/store/helm.go @@ -1222,62 +1222,111 @@ func InstallHelmChart(cfg *rest.Config, releaseName, namespace, chartName, repoU return nil } -// InstallHelmChartStream runs helm install asynchronously and pushes log lines to the returned channel. -// It waits for chart resources to become ready before sending "DONE", so the stream can stay open until -// helmInstallTimeout or until ctx is canceled. -// The channel is closed when the operation finishes; a final line of "ERROR: ", "ABORTED", or "DONE" signals the outcome. -// Cancel ctx to abort a waiting install (e.g. stuck waiting for PVCs). -func InstallHelmChartStream(ctx context.Context, cfg *rest.Config, releaseName, namespace, chartName, repoURL, version, valuesYAML string) <-chan string { +type HelmInstallLifecycle interface { + StartLoading() error + MarkInstalling() error + RecordLog(line string) error + Finish(installErr error) error +} + +// InstallHelmChartStream runs a Helm install independently of the browser +// request. Lifecycle persistence is supplied by the caller so store remains +// independent of the database layer. +func InstallHelmChartStream(ctx context.Context, lifecycle HelmInstallLifecycle, cfg *rest.Config, releaseName, namespace, chartName, repoURL, version, valuesYAML string) <-chan string { logCh := make(chan string, 64) + if lifecycle == nil { + logCh <- "ERROR: Helm install lifecycle is required" + close(logCh) + return logCh + } go func() { defer close(logCh) + streamCtx := ctx + if streamCtx == nil { + streamCtx = context.Background() + } + installCtx := context.WithoutCancel(streamCtx) send := func(line string) bool { + if err := lifecycle.RecordLog(line); err != nil { + logrus.Warnf("failed to persist Helm operation log: %v", err) + } select { case logCh <- line: return true - case <-ctx.Done(): + case <-streamCtx.Done(): return false } } + if err := lifecycle.StartLoading(); err != nil { + send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after loading error: %v", finishErr) + } + return + } logFn := func(format string, args ...interface{}) { send(fmt.Sprintf(format, args...)) } actionConfig, err := newHelmConfigWithLog(cfg, namespace, logFn) if err != nil { send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after configuration error: %v", finishErr) + } return } helmChart, err := loadChart(chartName, repoURL, version) if err != nil { send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after chart loading error: %v", finishErr) + } return } vals, err := parseValues(valuesYAML) if err != nil { send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after values parsing error: %v", finishErr) + } return } attachHelmCapabilities(actionConfig, cfg, namespace, logFn) + if err := lifecycle.MarkInstalling(); err != nil { + send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after phase transition error: %v", finishErr) + } + return + } install := action.NewInstall(actionConfig) install.ReleaseName = releaseName install.Namespace = namespace install.CreateNamespace = true install.Wait = true install.Timeout = helmInstallTimeout - if _, err = install.RunWithContext(ctx, helmChart, vals); err != nil { - if ctx.Err() != nil { - send("ABORTED") - } else { - for _, line := range helmReleaseDiagnostics(ctx, cfg, releaseName, namespace) { - if !send(line) { - return - } - } - send("ERROR: " + err.Error()) + if _, err = install.RunWithContext(installCtx, helmChart, vals); err != nil { + for _, line := range helmReleaseDiagnostics(installCtx, cfg, releaseName, namespace) { + send(line) + } + send("ERROR: " + err.Error()) + if finishErr := lifecycle.Finish(err); finishErr != nil { + logrus.Errorf("failed to finish Helm operation after install error: %v", finishErr) } return } - send("DONE") + if err := lifecycle.Finish(nil); err != nil { + logrus.Warnf("failed to finish Helm operation: %v", err) + select { + case logCh <- "ERROR: " + err.Error(): + case <-streamCtx.Done(): + } + return + } + select { + case logCh <- "DONE": + case <-streamCtx.Done(): + } }() return logCh } diff --git a/web/src/HelmInstallModal.js b/web/src/HelmInstallModal.js index 5ec3d7cb..02390f3a 100644 --- a/web/src/HelmInstallModal.js +++ b/web/src/HelmInstallModal.js @@ -3,8 +3,17 @@ import {Alert, Button, Form, Input, Modal, Select, Spin, Typography} from "antd" import {useTranslation} from "react-i18next"; import * as HelmBackend from "./backend/HelmBackend"; import * as NamespaceBackend from "./backend/NamespaceBackend"; +import { + findStoredHelmTask, + helmTaskMatchesIdentity, + helmTaskPollRetryDelay, + helmTaskStorageKey, + helmTaskStorageSchemaVersion, + removeStoredHelmTask +} from "./helmTaskStorage"; const {Text} = Typography; +const helmOperationTaskNotFoundCode = "helm_task_not_found"; export default function HelmInstallModal({open, chart, onClose, onInstalled}) { const {t} = useTranslation(); @@ -13,28 +22,168 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { const [valuesYAML, setValuesYAML] = useState(""); const [valuesLoading, setValuesLoading] = useState(false); const [installing, setInstalling] = useState(false); - const [aborted, setAborted] = useState(false); + const [pollingPaused, setPollingPaused] = useState(false); + const [activeTaskId, setActiveTaskId] = useState(null); const [done, setDone] = useState(false); const [error, setError] = useState(null); + const [storageWarning, setStorageWarning] = useState(null); const [logs, setLogs] = useState([]); const logEndRef = useRef(null); - const abortCtrlRef = useRef(null); + const taskIdRef = useRef(null); + const taskStorageKeyRef = useRef(null); + const taskIdentityRef = useRef(null); + const pollTimerRef = useRef(null); + const pollGenerationRef = useRef(0); + const streamAbortRef = useRef(null); + const mountedRef = useRef(true); + const submittingRef = useRef(false); + + const stopTaskPolling = () => { + pollGenerationRef.current += 1; + if (pollTimerRef.current) { + clearTimeout(pollTimerRef.current); + pollTimerRef.current = null; + } + }; + + const forgetTask = (storageKey = taskStorageKeyRef.current) => { + removeStoredHelmTask(storageKey); + if (!storageKey || taskStorageKeyRef.current === storageKey) { + taskIdRef.current = null; + setActiveTaskId(null); + taskStorageKeyRef.current = null; + taskIdentityRef.current = null; + } + }; + + const monitorTask = ( + taskId, + storageKey = taskStorageKeyRef.current, + expectedIdentity = taskIdentityRef.current + ) => { + if (!taskId) { + setInstalling(false); + setPollingPaused(false); + submittingRef.current = false; + return; + } + stopTaskPolling(); + setPollingPaused(false); + const generation = pollGenerationRef.current; + let consecutiveFailures = 0; + const poll = () => { + HelmBackend.getHelmOperationTask(taskId) + .then(res => { + if (!mountedRef.current || generation !== pollGenerationRef.current) {return;} + if (res.status !== "ok") { + if (res.data === helmOperationTaskNotFoundCode) { + forgetTask(storageKey); + submittingRef.current = false; + } else { + setPollingPaused(true); + submittingRef.current = true; + } + setError(res.msg); + setInstalling(false); + return; + } + consecutiveFailures = 0; + setError(null); + const task = res.data; + if (!task || !task.id || !task.status) { + setError(t("helm:Unable to refresh installation status: invalid response")); + setInstalling(false); + setPollingPaused(true); + submittingRef.current = true; + return; + } + const matchesExpectedTask = helmTaskMatchesIdentity(task, taskId, expectedIdentity); + if (!matchesExpectedTask) { + forgetTask(storageKey); + setError(t("helm:The saved installation task no longer matches this chart and was discarded")); + setInstalling(false); + setPollingPaused(false); + submittingRef.current = false; + return; + } + const taskLogs = Array.isArray(res.data2) ? res.data2 : []; + setLogs(taskLogs + .map(log => typeof log?.message === "string" ? log.message : "") + .filter(Boolean)); + if (task.status === "succeeded") { + setDone(true); + setInstalling(false); + setPollingPaused(false); + submittingRef.current = false; + forgetTask(storageKey); + return; + } + if (task.status === "failed") { + setError(task.errorMsg || t("helm:Helm operation failed")); + setInstalling(false); + setPollingPaused(false); + submittingRef.current = false; + forgetTask(storageKey); + return; + } + setInstalling(true); + pollTimerRef.current = setTimeout(poll, 2000); + }) + .catch(e => { + if (!mountedRef.current || generation !== pollGenerationRef.current) {return;} + consecutiveFailures += 1; + setError(t("helm:Unable to refresh installation status", {error: e.message})); + if (consecutiveFailures >= 6) { + setInstalling(false); + setPollingPaused(true); + submittingRef.current = true; + return; + } + setInstalling(true); + const retryDelay = helmTaskPollRetryDelay(consecutiveFailures); + pollTimerRef.current = setTimeout(poll, retryDelay); + }); + }; + poll(); + }; useEffect(() => { if (!open || !chart) {return;} setError(null); + setStorageWarning(null); setLogs([]); setDone(false); - setAborted(false); + setInstalling(false); + setPollingPaused(false); + taskIdRef.current = null; + setActiveTaskId(null); + taskStorageKeyRef.current = null; + taskIdentityRef.current = null; + submittingRef.current = false; + stopTaskPolling(); + streamAbortRef.current?.abort(); + streamAbortRef.current = null; + + const savedTask = findStoredHelmTask(chart.chartName); + if (savedTask) { + taskIdRef.current = savedTask.taskId; + setActiveTaskId(savedTask.taskId); + taskStorageKeyRef.current = savedTask.key; + taskIdentityRef.current = savedTask; + submittingRef.current = true; + setInstalling(true); + monitorTask(savedTask.taskId, savedTask.key, savedTask); + } NamespaceBackend.getNamespaces().then(res => { + if (!mountedRef.current) {return;} if (res.status === "ok") { const ns = res.data ?? []; setNamespaces(ns); const def = ns.find(n => n.name === "default") ? "default" : (ns[0]?.name ?? "default"); form.setFieldsValue({ - releaseName: chart.chartName, - namespace: def, + releaseName: savedTask?.releaseName || chart.chartName, + namespace: savedTask?.namespace || def, version: chart.version ?? "", }); } @@ -45,16 +194,29 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { setValuesYAML(""); HelmBackend.getHelmChartValues(chart.chartName, chart.repoURL, chart.version ?? "") .then(res => { + if (!mountedRef.current) {return;} if (res.status === "ok") { setValuesYAML(res.data ?? ""); } else { setError(res.msg); } }) - .finally(() => setValuesLoading(false)); + .finally(() => { + if (mountedRef.current) {setValuesLoading(false);} + }); } }, [open, chart, form]); + useEffect(() => { + mountedRef.current = true; + return () => { + mountedRef.current = false; + stopTaskPolling(); + streamAbortRef.current?.abort(); + streamAbortRef.current = null; + }; + }, []); + useEffect(() => { if (logEndRef.current) { logEndRef.current.scrollIntoView({behavior: "smooth"}); @@ -62,36 +224,42 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { }, [logs]); const handleClose = () => { + stopTaskPolling(); + streamAbortRef.current?.abort(); + streamAbortRef.current = null; + taskIdRef.current = null; + setActiveTaskId(null); + taskStorageKeyRef.current = null; + taskIdentityRef.current = null; + submittingRef.current = false; form.resetFields(); setValuesYAML(""); setError(null); + setStorageWarning(null); setLogs([]); setDone(false); - setAborted(false); setInstalling(false); + setPollingPaused(false); onClose(); }; - const handleAbort = () => { - if (abortCtrlRef.current) { - abortCtrlRef.current.abort(); - } - }; - const handleOk = () => { - if (done || aborted) { - if (done) {onInstalled?.();} + if (done) { + onInstalled?.(); handleClose(); return; } + if (submittingRef.current) {return;} + stopTaskPolling(); + submittingRef.current = true; form.validateFields().then(values => { setInstalling(true); - setAborted(false); + setPollingPaused(false); setError(null); setLogs([]); - - const ctrl = new AbortController(); - abortCtrlRef.current = ctrl; + const streamController = new AbortController(); + streamAbortRef.current?.abort(); + streamAbortRef.current = streamController; HelmBackend.installHelmChartStream( { @@ -103,34 +271,80 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { valuesYAML, }, line => { - if (line === "ABORTED") { - setAborted(true); + if (!mountedRef.current) {return;} + if (line.startsWith("TASK_ID:")) { + const taskId = line.slice("TASK_ID:".length).trim(); + const storageKey = helmTaskStorageKey(chart.chartName, values.namespace, values.releaseName); + taskIdRef.current = taskId; + setActiveTaskId(taskId); + taskStorageKeyRef.current = storageKey; + taskIdentityRef.current = { + chartName: chart.chartName, + namespace: values.namespace, + releaseName: values.releaseName, + }; + try { + window.localStorage.setItem(storageKey, JSON.stringify({ + schemaVersion: helmTaskStorageSchemaVersion, + taskId, + createdAt: Date.now(), + chartName: chart.chartName, + namespace: values.namespace, + releaseName: values.releaseName, + })); + } catch (_) { + setStorageWarning(t("helm:This browser cannot save the installation task for later recovery")); + } } else { setLogs(prev => [...prev, line]); } }, - ctrl.signal + streamController.signal ) .then(status => { + if (!mountedRef.current) {return;} if (status === "DONE") { setDone(true); + setInstalling(false); + setPollingPaused(false); + setStorageWarning(null); + submittingRef.current = false; + forgetTask(); } }) .catch(e => { - if (e.name === "AbortError") { - setAborted(true); - } else { - setError(e.message); + if (!mountedRef.current) {return;} + if (streamController.signal.aborted) {return;} + if (taskIdRef.current) { + monitorTask(taskIdRef.current, taskStorageKeyRef.current, taskIdentityRef.current); + return; } + setError(e.message); + setInstalling(false); + setPollingPaused(false); + submittingRef.current = false; }) - .finally(() => setInstalling(false)); + .finally(() => { + if (streamAbortRef.current === streamController) { + streamAbortRef.current = null; + } + }); + }).catch(() => { + submittingRef.current = false; }); }; if (!chart) {return null;} const nsOptions = namespaces.map(ns => ({label: ns.name, value: ns.name})); - const showLog = installing || done || aborted || (error && logs.length > 0); + const showLog = installing || pollingPaused || done || (error && logs.length > 0); + const hasActiveTask = Boolean(activeTaskId) && !done; + let closeLabel = t("general:Cancel"); + if (hasActiveTask) { + closeLabel = t("helm:Close and continue in background"); + } else if (done) { + closeLabel = t("general:Close"); + } const lineColor = (line, i, total) => { if (line.startsWith("ERROR")) {return "#f87171";} @@ -151,25 +365,28 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { } open={open} - onCancel={installing ? undefined : handleClose} - closable={!installing} - maskClosable={!installing} + onCancel={handleClose} + closable + maskClosable={false} footer={
- {installing && ( - - )} - {!installing && ( - - )} - {!done && !aborted && ( + + {!done && !pollingPaused && ( )} - {(done || aborted) && ( + {pollingPaused && ( + + )} + {done && ( @@ -182,8 +399,25 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { {error && ( setError(null)} /> )} - {aborted && ( - + + {storageWarning && ( + setStorageWarning(null)} + /> + )} + + {hasActiveTask && ( + )} {!showLog && ( @@ -240,10 +474,10 @@ export default function HelmInstallModal({open, chart, onClose, onInstalled}) { height: 340, overflowY: "auto", lineHeight: 1.6, }} > - {logs.length === 0 && installing && ( + {logs.length === 0 && (installing || pollingPaused) && ( - - {t("helm:Installing")}... + {installing && } + {pollingPaused ? t("helm:Status check paused") : `${t("helm:Installing")}...`} )} {logs.map((line, i) => ( diff --git a/web/src/HelmInstallModal.test.js b/web/src/HelmInstallModal.test.js new file mode 100644 index 00000000..ec55ddac --- /dev/null +++ b/web/src/HelmInstallModal.test.js @@ -0,0 +1,90 @@ +/* eslint-env jest */ + +import { + findStoredHelmTask, + helmTaskMatchesIdentity, + helmTaskPollRetryDelay, + helmTaskStorageKey +} from "./helmTaskStorage"; + +describe("Helm install task recovery", () => { + const now = 1_800_000_000_000; + + beforeEach(() => { + window.localStorage.clear(); + jest.spyOn(Date, "now").mockReturnValue(now); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + test("restores a fresh task with complete chart identity", () => { + const key = helmTaskStorageKey("demo", "apps", "demo-release"); + window.localStorage.setItem(key, JSON.stringify({ + schemaVersion: 1, + taskId: 42, + createdAt: now - 1000, + chartName: "demo", + namespace: "apps", + releaseName: "demo-release", + })); + + expect(findStoredHelmTask("demo")).toMatchObject({ + key, + taskId: "42", + chartName: "demo", + namespace: "apps", + releaseName: "demo-release", + }); + }); + + test("migrates the previous JSON format but rejects identity-free task ids", () => { + const legacyKey = helmTaskStorageKey("demo", "apps", "legacy-release"); + const rawKey = helmTaskStorageKey("demo", "apps", "raw-release"); + window.localStorage.setItem(legacyKey, JSON.stringify({ + taskId: 43, + createdAt: now - 1000, + namespace: "apps", + releaseName: "legacy-release", + })); + window.localStorage.setItem(rawKey, "44"); + + expect(findStoredHelmTask("demo")).toMatchObject({ + key: legacyKey, + taskId: "43", + chartName: "demo", + }); + expect(window.localStorage.getItem(rawKey)).toBeNull(); + }); + + test("discards expired tasks before attempting recovery", () => { + const key = helmTaskStorageKey("demo", "apps", "expired-release"); + window.localStorage.setItem(key, JSON.stringify({ + schemaVersion: 1, + taskId: 45, + createdAt: now - (25 * 60 * 60 * 1000), + chartName: "demo", + namespace: "apps", + releaseName: "expired-release", + })); + + expect(findStoredHelmTask("demo")).toBeNull(); + expect(window.localStorage.getItem(key)).toBeNull(); + }); + + test("matches a recovered task against its complete identity", () => { + const task = {id: 46, chartName: "demo", namespace: "apps", releaseName: "demo-release"}; + const identity = {chartName: "demo", namespace: "apps", releaseName: "demo-release"}; + + expect(helmTaskMatchesIdentity(task, "46", identity)).toBe(true); + expect(helmTaskMatchesIdentity({...task, namespace: "other"}, "46", identity)).toBe(false); + expect(helmTaskMatchesIdentity(task, "47", identity)).toBe(false); + }); + + test("backs off failed status polls with a bounded delay", () => { + expect(helmTaskPollRetryDelay(1)).toBe(2000); + expect(helmTaskPollRetryDelay(2)).toBe(4000); + expect(helmTaskPollRetryDelay(10)).toBe(30000); + }); +}); diff --git a/web/src/backend/HelmBackend.js b/web/src/backend/HelmBackend.js index 8e77d975..78259a55 100644 --- a/web/src/backend/HelmBackend.js +++ b/web/src/backend/HelmBackend.js @@ -44,6 +44,12 @@ export function getHelmReleases(namespace = "all") { }).then(r => r.json()); } +export function getHelmOperationTask(id) { + return fetch(`${Setting.ServerUrl}/api/get-helm-operation-task?id=${encodeURIComponent(id)}`, { + credentials: "include", headers: lang(), + }).then(r => r.json()); +} + export function installHelmChart(payload) { return fetch(`${Setting.ServerUrl}/api/install-helm-chart`, { method: "POST", credentials: "include", headers: jsonHeaders(), body: JSON.stringify(payload), @@ -51,8 +57,8 @@ export function installHelmChart(payload) { } // installHelmChartStream posts the payload then reads the SSE response line-by-line. -// onLine(line) is called for each log line; returns "DONE" or "ABORTED", and rejects on "ERROR: ...". -// Pass an AbortSignal to cancel mid-install (sends abort to the server, stops the wait loop). +// onLine(line) is called for each log line; returns "DONE" and rejects on "ERROR: ...". +// Closing the browser stream does not cancel a submitted install. export async function installHelmChartStream(payload, onLine, signal) { const resp = await fetch(`${Setting.ServerUrl}/api/install-helm-chart-stream`, { method: "POST", credentials: "include", headers: jsonHeaders(), body: JSON.stringify(payload), signal, @@ -71,7 +77,7 @@ export async function installHelmChartStream(payload, onLine, signal) { if (line) { onLine(line); if (line.startsWith("ERROR: ")) {throw new Error(line.slice(7));} - if (line === "DONE" || line === "ABORTED") {return line;} + if (line === "DONE") {return line;} } } } diff --git a/web/src/backend/HelmBackend.test.js b/web/src/backend/HelmBackend.test.js index a213b33a..a3421d97 100644 --- a/web/src/backend/HelmBackend.test.js +++ b/web/src/backend/HelmBackend.test.js @@ -37,19 +37,19 @@ describe("installHelmChartStream", () => { jest.resetAllMocks(); }); - test("returns ABORTED when the server aborts the install stream", async() => { + test("rejects when the server reports an install failure", async() => { global.fetch = jest.fn().mockResolvedValue(mockStreamResponse([ "data: creating 1 resource(s)\n\n", - "data: ABORTED\n\n", + "data: ERROR: install failed\n\n", ])); const onLine = jest.fn(); - const status = await installHelmChartStream({releaseName: "demo"}, onLine); + await expect(installHelmChartStream({releaseName: "demo"}, onLine)) + .rejects.toThrow("install failed"); - expect(status).toBe("ABORTED"); expect(onLine).toHaveBeenCalledTimes(2); expect(onLine).toHaveBeenNthCalledWith(1, "creating 1 resource(s)"); - expect(onLine).toHaveBeenNthCalledWith(2, "ABORTED"); + expect(onLine).toHaveBeenNthCalledWith(2, "ERROR: install failed"); }); test("returns DONE when the server completes the install stream", async() => { @@ -65,4 +65,19 @@ describe("installHelmChartStream", () => { expect(onLine).toHaveBeenCalledTimes(2); expect(onLine).toHaveBeenNthCalledWith(2, "DONE"); }); + + test("forwards the task id and abort signal", async() => { + global.fetch = jest.fn().mockResolvedValue(mockStreamResponse([ + "data: TASK_ID:42\n\n", + "data: DONE\n\n", + ])); + const signal = {aborted: false}; + const onLine = jest.fn(); + + await installHelmChartStream({releaseName: "demo"}, onLine, signal); + + expect(global.fetch.mock.calls[0][1].signal).toBe(signal); + expect(onLine).toHaveBeenNthCalledWith(1, "TASK_ID:42"); + expect(onLine).toHaveBeenNthCalledWith(2, "DONE"); + }); }); diff --git a/web/src/helmTaskStorage.js b/web/src/helmTaskStorage.js new file mode 100644 index 00000000..31e8d05a --- /dev/null +++ b/web/src/helmTaskStorage.js @@ -0,0 +1,70 @@ +export const helmTaskStorageSchemaVersion = 1; + +const helmTaskStorageMaxAgeMs = 24 * 60 * 60 * 1000; +const helmTaskStoragePrefix = chartName => `casos.helmTask.${encodeURIComponent(chartName)}.`; + +export const helmTaskStorageKey = (chartName, namespace, releaseName) => + `${helmTaskStoragePrefix(chartName)}${encodeURIComponent(namespace)}.${encodeURIComponent(releaseName)}`; + +export const removeStoredHelmTask = key => { + if (!key) {return;} + try { + window.localStorage.removeItem(key); + } catch (_) { + // Storage may be unavailable; task polling still works for this session. + } +}; + +export const helmTaskMatchesIdentity = (task, taskId, expectedIdentity) => Boolean( + task && expectedIdentity && + String(task.id) === String(taskId) && + task.chartName === expectedIdentity.chartName && + task.namespace === expectedIdentity.namespace && + task.releaseName === expectedIdentity.releaseName +); + +export const helmTaskPollRetryDelay = consecutiveFailures => + Math.min(2000 * (2 ** Math.max(consecutiveFailures - 1, 0)), 30000); + +export const findStoredHelmTask = chartName => { + const prefix = helmTaskStoragePrefix(chartName); + const matches = []; + const invalidKeys = []; + try { + for (let i = 0; i < window.localStorage.length; i += 1) { + const key = window.localStorage.key(i); + if (!key?.startsWith(prefix)) {continue;} + const raw = window.localStorage.getItem(key); + try { + const stored = JSON.parse(raw); + const createdAt = Number(stored?.createdAt); + const isFresh = createdAt > Date.now() - helmTaskStorageMaxAgeMs; + const hasTaskIdentity = /^\d+$/.test(String(stored?.taskId ?? "")) && + typeof stored?.namespace === "string" && stored.namespace && + typeof stored?.releaseName === "string" && stored.releaseName; + const isCurrentSchema = stored?.schemaVersion === helmTaskStorageSchemaVersion && + stored.chartName === chartName; + const isLegacySchema = stored?.schemaVersion === undefined; + if (hasTaskIdentity && (isCurrentSchema || isLegacySchema) && isFresh) { + matches.push({ + key, + taskId: String(stored.taskId), + createdAt, + chartName, + namespace: stored.namespace, + releaseName: stored.releaseName, + }); + } else { + invalidKeys.push(key); + } + } catch (_) { + invalidKeys.push(key); + } + } + } catch (_) { + return null; + } + invalidKeys.forEach(removeStoredHelmTask); + matches.sort((a, b) => b.createdAt - a.createdAt); + return matches[0] ?? null; +}; diff --git a/web/src/locales/en/data.json b/web/src/locales/en/data.json index 0d4717d0..2e7fee9e 100644 --- a/web/src/locales/en/data.json +++ b/web/src/locales/en/data.json @@ -101,8 +101,15 @@ "Version": "Version", "Values (YAML)": "Values (YAML)", "Loading values": "Loading default values", - "Abort": "Abort", - "Install aborted": "Install aborted — resources may have been partially created" + "Close and continue in background": "Close and continue in background", + "Closing this window does not cancel the installation": "Closing this window does not cancel the installation", + "Retry status check": "Retry status check", + "Status check paused": "Status check paused", + "This browser cannot save the installation task for later recovery": "This browser cannot save the installation task for later recovery", + "Unable to refresh installation status": "Unable to refresh installation status: {{error}}", + "Unable to refresh installation status: invalid response": "Unable to refresh installation status: invalid response", + "The saved installation task no longer matches this chart and was discarded": "The saved installation task no longer matches this chart and was discarded.", + "Helm operation failed": "Helm operation failed" }, "monitor": { "Abnormal Pods": "Abnormal Pods", diff --git a/web/src/locales/zh/data.json b/web/src/locales/zh/data.json index 2f38679a..781eeb4b 100644 --- a/web/src/locales/zh/data.json +++ b/web/src/locales/zh/data.json @@ -101,8 +101,15 @@ "Version": "版本", "Values (YAML)": "Values (YAML)", "Loading values": "加载默认配置中", - "Abort": "中止", - "Install aborted": "安装已中止 — 部分资源可能已创建" + "Close and continue in background": "关闭并在后台继续", + "Closing this window does not cancel the installation": "关闭此窗口不会取消安装", + "Retry status check": "重试状态检查", + "Status check paused": "状态检查已暂停", + "This browser cannot save the installation task for later recovery": "此浏览器无法保存安装任务以供稍后恢复", + "Unable to refresh installation status": "无法刷新安装状态：{{error}}", + "Unable to refresh installation status: invalid response": "无法刷新安装状态：响应无效", + "The saved installation task no longer matches this chart and was discarded": "已保存的安装任务与当前 chart 不再匹配，已丢弃。", + "Helm operation failed": "Helm 操作失败" }, "monitor": { "Abnormal Pods": "异常 Pod",