feat: validate worker operational readiness

.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() {

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 {

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 {