feat: persist helm install lifecycle

.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

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,27 +239,120 @@ 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")
 	w.Header().Set("Cache-Control", "no-cache")
 	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() {