feat: remove test mode (X-Test-Sub), all functions stored in S3 via storagesvc

- Remove FISSION_TEST_MODE and X-Test-Sub bypass from authMiddleware - Remove testMode field from Server/Config structs - Remove TestAuthenticator branch from main.go - Add uploadToStoragesvc/buildDeploySpec: all function code uploaded to S3 - Add deleteFromStoragesvc: archive deleted from S3 on function delete - Add FISSION_STORAGESVC_URL env to main.go and deploy/console.yaml - image: naeel/fission-console:v1.3.31
2026-05-03 13:37:12 +04:00
parent e8f06d4ab2
commit a09291e208
5 changed files with 141 additions and 40 deletions
@@ -11,6 +11,7 @@ import (
 	"log"
 	"net"
 	"net/http"
+	"net/url"
 	"regexp"
 	"strconv"
 	"strings"
@@ -51,6 +52,102 @@ func buildDeployArchive(lang, code string) ([]byte, error) {
 	}
 }

+// deleteFromStoragesvc удаляет архив из S3 через storagesvc по URL из Package spec.
+// URL имеет формат: http://storagesvc.../v1/archive?id=fission/UUID
+// Best-effort: ошибка логируется, но не прерывает операцию удаления.
+func (s *Server) deleteFromStoragesvc(ctx context.Context, archiveURL string) {
+	if s.storagesvcURL == "" || archiveURL == "" {
+		return
+	}
+	// archiveURL = "http://storagesvc.../v1/archive?id=fission/UUID"
+	// Строим DELETE URL к storagesvc, сохраняя query-параметр id
+	parsed, err := url.Parse(archiveURL)
+	if err != nil {
+		log.Printf("deleteFromStoragesvc: parse url %q: %v", archiveURL, err)
+		return
+	}
+	deleteURL := strings.TrimRight(s.storagesvcURL, "/") + "/v1/archive?" + parsed.RawQuery
+	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, deleteURL, nil)
+	if err != nil {
+		log.Printf("deleteFromStoragesvc: build request: %v", err)
+		return
+	}
+	resp, err := s.http.Do(req)
+	if err != nil {
+		log.Printf("deleteFromStoragesvc: %v", err)
+		return
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
+		body, _ := io.ReadAll(resp.Body)
+		log.Printf("deleteFromStoragesvc: status %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+		return
+	}
+	log.Printf("deleteFromStoragesvc: deleted %s", parsed.Query().Get("id"))
+}
+
+// uploadToStoragesvc загружает байты в Fission storagesvc и возвращает URL для package archive.
+// Если storagesvcURL не задан — возвращает пустую строку (fallback на literal).
+func (s *Server) uploadToStoragesvc(ctx context.Context, data []byte) (string, error) {
+	if s.storagesvcURL == "" {
+		log.Printf("uploadToStoragesvc: storagesvcURL is empty, skip upload")
+		return "", nil
+	}
+	log.Printf("uploadToStoragesvc: uploading %d bytes to %s", len(data), s.storagesvcURL)
+	uploadURL := strings.TrimRight(s.storagesvcURL, "/") + "/v1/archive"
+	body := &bytes.Reader{}
+	// multipart/form-data с полем uploadfile
+	var buf bytes.Buffer
+	boundary := fmt.Sprintf("fission%d", time.Now().UnixNano())
+	buf.WriteString("--" + boundary + "\r\n")
+	buf.WriteString(fmt.Sprintf("Content-Disposition: form-data; name=\"uploadfile\"; filename=\"archive.zip\"\r\n"))
+	buf.WriteString("Content-Type: application/octet-stream\r\n\r\n")
+	buf.Write(data)
+	buf.WriteString("\r\n--" + boundary + "--\r\n")
+	body = bytes.NewReader(buf.Bytes())
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, body)
+	if err != nil {
+		return "", fmt.Errorf("build storagesvc upload request: %w", err)
+	}
+	req.Header.Set("Content-Type", "multipart/form-data; boundary="+boundary)
+	req.Header.Set("X-File-Size", fmt.Sprintf("%d", len(data)))
+
+	resp, err := s.http.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("storagesvc upload: %w", err)
+	}
+	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
+		return "", fmt.Errorf("storagesvc upload status %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody)))
+	}
+	var result struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(respBody, &result); err != nil || result.ID == "" {
+		return "", fmt.Errorf("storagesvc upload: bad response: %s", string(respBody))
+	}
+	archiveURL := strings.TrimRight(s.storagesvcURL, "/") + "/v1/archive?id=" + result.ID
+	return archiveURL, nil
+}
+
+// buildDeploySpec строит spec.deployment для Fission Package.
+// Если storagesvcURL задан — загружает архив в S3 через storagesvc и возвращает type: url.
+// Иначе — возвращает type: literal с base64-кодом.
+func (s *Server) buildDeploySpec(ctx context.Context, data []byte) (map[string]any, error) {
+	archiveURL, err := s.uploadToStoragesvc(ctx, data)
+	if err != nil {
+		log.Printf("storagesvc upload failed, falling back to literal: %v", err)
+		// fallback — сохраняем как literal
+		return map[string]any{"type": "literal", "literal": base64.StdEncoding.EncodeToString(data)}, nil
+	}
+	if archiveURL == "" {
+		return map[string]any{"type": "literal", "literal": base64.StdEncoding.EncodeToString(data)}, nil
+	}
+	return map[string]any{"type": "url", "url": archiveURL}, nil
+}
+
 func (s *Server) resolveInvokeTimeout(fn *unstructured.Unstructured) time.Duration {
 	if fn != nil {
 		seconds, found, err := unstructured.NestedInt64(fn.Object, "spec", "functionTimeout")
@@ -313,7 +410,7 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
 	// Строим Package spec в зависимости от языка:
 	// - Go: source package → builder job компилирует в .so плагин
 	// - Node.js: deployment zip с ESM wrapper (package.json + main.js)
-	// - Остальные: deployment literal с кодом напрямую
+	// - Остальные: deployment archive с кодом (S3 или literal fallback)
 	var pkgSpec map[string]any
 	if req.Language == "go" {
 		srcZip, err := runtime.BuildGoSourceZip(req.Code)
@@ -321,12 +418,13 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
 			writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("build go source archive: %v", err))
 			return
 		}
-		literal := base64.StdEncoding.EncodeToString(srcZip)
+		srcSpec, srcErr := s.buildDeploySpec(ctx, srcZip)
+		if srcErr != nil {
+			writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("upload go source: %v", srcErr))
+			return
+		}
 		pkgSpec = map[string]any{
-			"source": map[string]any{
-				"type":    "literal",
-				"literal": literal,
-			},
+			"source":       srcSpec,
 			"deployment":   map[string]any{},
 			"environment":  map[string]any{"name": req.Environment, "namespace": ns},
 			"buildcommand": "build",
@@ -337,8 +435,13 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
 			writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("build %s archive: %v", req.Language, archiveErr))
 			return
 		}
+		deploySpec, uploadErr := s.buildDeploySpec(ctx, deployBytes)
+		if uploadErr != nil {
+			writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("upload %s archive: %v", req.Language, uploadErr))
+			return
+		}
 		pkgSpec = map[string]any{
-			"deployment":  map[string]any{"type": "literal", "literal": base64.StdEncoding.EncodeToString(deployBytes)},
+			"deployment":  deploySpec,
 			"environment": map[string]any{"name": req.Environment, "namespace": ns},
 			"source":      map[string]any{},
 		}
@@ -563,12 +666,17 @@ func (s *Server) handleUpdateFunctionCode(w http.ResponseWriter, r *http.Request
 	}()
 	now := time.Now().UTC()
 	newPkgName := name + "-pkg-" + strconv.FormatInt(time.Now().UnixMilli(), 36)
+	deploySpec, uploadErr := s.buildDeploySpec(ctx, deployBytes)
+	if uploadErr != nil {
+		writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("upload %s archive: %v", lang, uploadErr))
+		return
+	}
 	newPkg := &unstructured.Unstructured{Object: map[string]any{
 		"apiVersion": "fission.io/v1",
 		"kind":       "Package",
 		"metadata":   map[string]any{"name": newPkgName, "namespace": ns},
 		"spec": map[string]any{
-			"deployment":  map[string]any{"type": "literal", "literal": base64.StdEncoding.EncodeToString(deployBytes)},
+			"deployment":  deploySpec,
 			"environment": map[string]any{"name": envName, "namespace": ns},
 			"source":      map[string]any{},
 		},
@@ -1000,10 +1108,22 @@ func (s *Server) handleDeleteFunction(w http.ResponseWriter, r *http.Request, na
 	}

 	if pkgName != "" {
+		// Получаем URL архива из Package spec.deployment перед удалением, чтобы потом очистить S3
+		var archiveURL string
+		if pkg, pkgErr := s.dyn.Resource(fission.PackageGVR).Namespace(ns).Get(ctx, pkgName, metav1.GetOptions{}); pkgErr == nil {
+			deployType, _, _ := unstructured.NestedString(pkg.Object, "spec", "deployment", "type")
+			if deployType == "url" {
+				archiveURL, _, _ = unstructured.NestedString(pkg.Object, "spec", "deployment", "url")
+			}
+		}
 		if err := s.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, pkgName, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
 			writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("delete package %q: %v", pkgName, err))
 			return
 		}
+		// Удаляем архив из S3 после успешного удаления Package (best-effort)
+		if archiveURL != "" {
+			go s.deleteFromStoragesvc(context.Background(), archiveURL)
+		}
 	}

 	// Убираем environment pool pods если язык больше не используется (best-effort)