Nail/fission-console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Organize LLM assets and function lifecycle fixes

Browse Source
This commit is contained in:
Naeel
2026-04-28 13:04:08 +03:00
parent 82eba079f6
commit a534fddd2c
22 changed files with 1409 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
console/internal/api/function_timestamps.go
+70
View File
@@ -0,0 +1,70 @@
package api
import (
"fmt"
"time"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
)
const (
functionCreatedAtAnnotation = "fission-console/created-at"
functionUpdatedAtAnnotation = "fission-console/updated-at"
)
func ensureFunctionTimestamps(fn *unstructured.Unstructured, now time.Time) {
if fn == nil {
return
}
annotations := fn.GetAnnotations()
if annotations == nil {
annotations = map[string]string{}
}
if annotations[functionCreatedAtAnnotation] == "" {
if created := fn.GetCreationTimestamp(); !created.IsZero() {
annotations[functionCreatedAtAnnotation] = created.UTC().Format(time.RFC3339)
} else {
annotations[functionCreatedAtAnnotation] = now.UTC().Format(time.RFC3339)
}
}
annotations[functionUpdatedAtAnnotation] = now.UTC().Format(time.RFC3339)
fn.SetAnnotations(annotations)
}
func functionTimestampResponse(fn *unstructured.Unstructured) map[string]string {
result := map[string]string{}
if fn == nil {
return result
}
annotations := fn.GetAnnotations()
if annotations == nil {
annotations = map[string]string{}
}
if created := annotations[functionCreatedAtAnnotation]; created != "" {
result["created_at"] = created
} else if ts := fn.GetCreationTimestamp(); !ts.IsZero() {
result["created_at"] = ts.UTC().Format(time.RFC3339)
}
if updated := annotations[functionUpdatedAtAnnotation]; updated != "" {
result["updated_at"] = updated
} else if v := result["created_at"]; v != "" {
result["updated_at"] = v
}
return result
}
func formatRFC3339Now(now time.Time) string {
return now.UTC().Format(time.RFC3339)
}
func parseRFC3339(value string) (time.Time, error) {
return time.Parse(time.RFC3339, value)
}
func mustParseRFC3339(value string) time.Time {
parsed, err := parseRFC3339(value)
if err != nil {
panic(fmt.Sprintf("parse RFC3339 %q: %v", value, err))
}
return parsed
}
console/internal/api/function_timestamps_test.go
+88
View File
@@ -0,0 +1,88 @@
package api
import (
"testing"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
)
func TestEnsureFunctionTimestampsSetsBothOnCreate(t *testing.T) {
fn := &unstructured.Unstructured{}
now := mustParseRFC3339("2026-04-28T10:11:12Z")
ensureFunctionTimestamps(fn, now)
ann := fn.GetAnnotations()
if ann[functionCreatedAtAnnotation] != now.Format(time.RFC3339) {
t.Fatalf("created_at = %q, want %q", ann[functionCreatedAtAnnotation], now.Format(time.RFC3339))
}
if ann[functionUpdatedAtAnnotation] != now.Format(time.RFC3339) {
t.Fatalf("updated_at = %q, want %q", ann[functionUpdatedAtAnnotation], now.Format(time.RFC3339))
}
}
func TestEnsureFunctionTimestampsPreservesExistingCreatedAt(t *testing.T) {
fn := &unstructured.Unstructured{}
created := "2026-04-27T09:10:11Z"
annotations := map[string]string{functionCreatedAtAnnotation: created}
fn.SetAnnotations(annotations)
now := mustParseRFC3339("2026-04-28T10:11:12Z")
ensureFunctionTimestamps(fn, now)
ann := fn.GetAnnotations()
if ann[functionCreatedAtAnnotation] != created {
t.Fatalf("created_at = %q, want %q", ann[functionCreatedAtAnnotation], created)
}
if ann[functionUpdatedAtAnnotation] != now.Format(time.RFC3339) {
t.Fatalf("updated_at = %q, want %q", ann[functionUpdatedAtAnnotation], now.Format(time.RFC3339))
}
}
func TestEnsureFunctionTimestampsBackfillsCreatedAtFromCreationTimestamp(t *testing.T) {
fn := &unstructured.Unstructured{}
fn.SetCreationTimestamp(metav1.NewTime(mustParseRFC3339("2026-04-26T08:09:10Z")))
now := mustParseRFC3339("2026-04-28T10:11:12Z")
ensureFunctionTimestamps(fn, now)
ann := fn.GetAnnotations()
if ann[functionCreatedAtAnnotation] != "2026-04-26T08:09:10Z" {
t.Fatalf("created_at = %q", ann[functionCreatedAtAnnotation])
}
if ann[functionUpdatedAtAnnotation] != now.Format(time.RFC3339) {
t.Fatalf("updated_at = %q, want %q", ann[functionUpdatedAtAnnotation], now.Format(time.RFC3339))
}
}
func TestFunctionTimestampResponse(t *testing.T) {
fn := &unstructured.Unstructured{}
annotations := map[string]string{
functionCreatedAtAnnotation: "2026-04-26T08:09:10Z",
functionUpdatedAtAnnotation: "2026-04-28T10:11:12Z",
}
fn.SetAnnotations(annotations)
resp := functionTimestampResponse(fn)
if resp["created_at"] != "2026-04-26T08:09:10Z" {
t.Fatalf("created_at = %q", resp["created_at"])
}
if resp["updated_at"] != "2026-04-28T10:11:12Z" {
t.Fatalf("updated_at = %q", resp["updated_at"])
}
}
func TestFunctionTimestampResponseFallsBackToCreationTimestamp(t *testing.T) {
fn := &unstructured.Unstructured{}
fn.SetCreationTimestamp(metav1.NewTime(mustParseRFC3339("2026-04-26T08:09:10Z")))
resp := functionTimestampResponse(fn)
if resp["created_at"] != "2026-04-26T08:09:10Z" {
t.Fatalf("created_at = %q", resp["created_at"])
}
if resp["updated_at"] != "2026-04-26T08:09:10Z" {
t.Fatalf("updated_at = %q", resp["updated_at"])
}
}
console/internal/api/handlers.go
+71 -37
View File
@@ -354,6 +354,9 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
fnAnnotations := map[string]any{
"fission-console/language": req.Language,
}
now := time.Now().UTC()
fnAnnotations[functionCreatedAtAnnotation] = now.Format(time.RFC3339)
fnAnnotations[functionUpdatedAtAnnotation] = now.Format(time.RFC3339)
if req.TTL != "" {
expiresAt, ttlErr := parseTTL(req.TTL)
if ttlErr != nil {
@@ -381,7 +384,7 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
"kind": "Function",
"metadata": map[string]any{"name": req.Name, "namespace": ns, "annotations": fnAnnotations},
"spec": map[string]any{
"environment": map[string]any{"name": req.Environment, "namespace": ns},
"environment": map[string]any{"name": req.Environment, "namespace": ns},
"functionTimeout": req.Timeout,
"InvokeStrategy": map[string]any{
"ExecutionStrategy": map[string]any{"ExecutorType": "poolmgr"},
@@ -490,6 +493,8 @@ func (s *Server) handleGetFunction(w http.ResponseWriter, r *http.Request, name
"package": packageName,
"entrypoint": entrypoint,
"timeout": functionTimeout,
"created_at": functionTimestampResponse(fn)["created_at"],
"updated_at": functionTimestampResponse(fn)["updated_at"],
"code": code,
"route": route,
"methods": methods,
@@ -498,11 +503,9 @@ func (s *Server) handleGetFunction(w http.ResponseWriter, r *http.Request, name
}
// handleUpdateFunctionCode обновляет код уже существующей функции.
// Обновляет Package.spec.deployment.literal и синхронизирует resourceVersion в Function.
//
// Почему нужно обновлять resourceVersion в Function:
// Fission executor кэширует Package по resourceVersion. Без обновления в Function
// executor будет использовать старый код до перезапуска pod.
// Создаёт НОВЫЙ Package (вместо обновления старого) чтобы executor сбросил кэш:
// executor кэширует function service по functionUid и не видит изменений в том же Package.
// Новое имя пакета гарантирует cache miss в executor.
func (s *Server) handleUpdateFunctionCode(w http.ResponseWriter, r *http.Request, name string) {
var req model.UpdateCodeRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
@@ -529,21 +532,7 @@ func (s *Server) handleUpdateFunctionCode(w http.ResponseWriter, r *http.Request
return
}
pkgName, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "packageref", "name")
if pkgName == "" {
writeJSONError(w, http.StatusBadGateway, "function has no package reference")
return
}
pkg, err := s.dyn.Resource(fission.PackageGVR).Namespace(ns).Get(ctx, pkgName, metav1.GetOptions{})
if err != nil {
status := http.StatusBadGateway
if apierrors.IsNotFound(err) {
status = http.StatusNotFound
}
writeJSONError(w, status, fmt.Sprintf("get package %q: %v", pkgName, err))
return
}
oldPkgName, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "packageref", "name")
// Определяем язык из аннотации — нужен для правильной упаковки
lang, _, _ := unstructured.NestedString(fn.Object, "metadata", "annotations", "fission-console/language")
@@ -553,38 +542,83 @@ func (s *Server) handleUpdateFunctionCode(w http.ResponseWriter, r *http.Request
return
}
literal := base64.StdEncoding.EncodeToString(deployBytes)
if err := unstructured.SetNestedField(pkg.Object, literal, "spec", "deployment", "literal"); err != nil {
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("set package literal: %v", err))
return
}
// Создаём новый Package с уникальным именем.
// Это единственный способ сбросить кэш executor: он кэширует по functionUid и
// не замечает изменений в существующем Package.
envName, _, _ := unstructured.NestedString(fn.Object, "spec", "environment", "name")
createdAt := func() time.Time {
ann := fn.GetAnnotations()
if ann != nil {
if v := strings.TrimSpace(ann[functionCreatedAtAnnotation]); v != "" {
if ts, err := parseRFC3339(v); err == nil {
return ts.UTC()
}
}
}
if ts := fn.GetCreationTimestamp(); !ts.IsZero() {
return ts.UTC()
}
return time.Time{}
}()
now := time.Now().UTC()
newPkgName := name + "-pkg-" + strconv.FormatInt(time.Now().UnixMilli(), 36)
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)},
"environment": map[string]any{"name": envName, "namespace": ns},
"source": map[string]any{},
},
}}
updatedPkg, err := s.dyn.Resource(fission.PackageGVR).Namespace(ns).Update(ctx, pkg, metav1.UpdateOptions{})
createdPkg, err := s.dyn.Resource(fission.PackageGVR).Namespace(ns).Create(ctx, newPkg, metav1.CreateOptions{})
if err != nil {
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("update package %q: %v", pkgName, err))
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("create new package: %v", err))
return
}
// Обновляем Function на новый Package
if err := unstructured.SetNestedField(fn.Object, normalizeFunctionTimeout(req.Timeout), "spec", "functionTimeout"); err != nil {
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("set function timeout: %v", err))
_ = s.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, newPkgName, metav1.DeleteOptions{})
return
}
// Синхронизируем resourceVersion в Function.spec.package.packageref
// Это триггерит executor перезагрузить код в pool pod
if err := unstructured.SetNestedField(fn.Object, updatedPkg.GetResourceVersion(), "spec", "package", "packageref", "resourceversion"); err != nil {
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("set function package resourceversion: %v", err))
ensureFunctionTimestamps(fn, now)
if createdAt.IsZero() {
createdAt = now
}
fnAnnotations := fn.GetAnnotations()
if fnAnnotations == nil {
fnAnnotations = map[string]string{}
}
fnAnnotations[functionCreatedAtAnnotation] = createdAt.UTC().Format(time.RFC3339)
fnAnnotations[functionUpdatedAtAnnotation] = now.Format(time.RFC3339)
fn.SetAnnotations(fnAnnotations)
if err := unstructured.SetNestedField(fn.Object, map[string]any{
"name": newPkgName,
"namespace": ns,
"resourceversion": createdPkg.GetResourceVersion(),
}, "spec", "package", "packageref"); err != nil {
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("set function packageref: %v", err))
_ = s.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, newPkgName, metav1.DeleteOptions{})
return
}
if _, err := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Update(ctx, fn, metav1.UpdateOptions{}); err != nil {
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("update function %q package ref: %v", name, err))
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("update function %q: %v", name, err))
_ = s.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, newPkgName, metav1.DeleteOptions{})
return
}
// Удаляем старый Package (best effort)
if oldPkgName != "" && oldPkgName != newPkgName {
_ = s.dyn.Resource(fission.PackageGVR).Namespace(ns).Delete(ctx, oldPkgName, metav1.DeleteOptions{})
}
writeAnyJSON(w, http.StatusOK, map[string]any{
"updated": true,
"package": pkgName,
"package_resourceversion": updatedPkg.GetResourceVersion(),
"updated": true,
"package": newPkgName,
})
}
console/internal/api/package.go
+29
View File
@@ -5,8 +5,10 @@ import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"io"
"net/http"
"regexp"
"sort"
"strings"
"unicode/utf8"
@@ -90,6 +92,9 @@ func decodeArchiveBytesToSource(decoded []byte) (string, error) {
if len(decoded) == 0 {
return "", io.ErrUnexpectedEOF
}
if src, ok := decodeNodeJSWrapperSource(decoded); ok {
return src, nil
}
// Если байты — валидный UTF-8, возвращаем напрямую (python, ruby, php)
if utf8.Valid(decoded) {
return string(decoded), nil
@@ -103,6 +108,24 @@ func decodeArchiveBytesToSource(decoded []byte) (string, error) {
return "", io.ErrUnexpectedEOF
}
var nodeJSWrapperPattern = regexp.MustCompile(`(?s)^const __mod = \{ exports: \{\} \};\s*\(new Function\('module', 'exports', (.+?)\)\)\(__mod, __mod\.exports\);\s*const _fn = __mod\.exports;`)
func decodeNodeJSWrapperSource(decoded []byte) (string, bool) {
text := string(decoded)
matches := nodeJSWrapperPattern.FindStringSubmatch(text)
if len(matches) != 2 {
return "", false
}
var code string
if err := json.Unmarshal([]byte(matches[1]), &code); err != nil {
return "", false
}
if strings.TrimSpace(code) == "" {
return "", false
}
return code, true
}
// decodeZipSource извлекает исходный UTF-8 файл из zip архива.
// Предпочитает хорошо известные имена файлов (main.py, handler.go и т.д.).
func decodeZipSource(zipBytes []byte) (string, error) {
@@ -118,6 +141,9 @@ func decodeZipSource(zipBytes []byte) (string, error) {
if strings.EqualFold(file.Name, name) {
content, readErr := readZipFile(file)
if readErr == nil && utf8.Valid(content) {
if src, ok := decodeNodeJSWrapperSource(content); ok {
return src, nil
}
return string(content), nil
}
}
@@ -136,6 +162,9 @@ func decodeZipSource(zipBytes []byte) (string, error) {
for _, file := range files {
content, readErr := readZipFile(file)
if readErr == nil && utf8.Valid(content) {
if src, ok := decodeNodeJSWrapperSource(content); ok {
return src, nil
}
return string(content), nil
}
}
console/internal/api/package_test.go
+98
View File
@@ -0,0 +1,98 @@
package api
import (
"fission-console/internal/runtime"
"testing"
)
func TestDecodeNodeJSWrapperSource(t *testing.T) {
wrapper := []byte("const __mod = { exports: {} };\n(new Function('module', 'exports', \"module.exports = async function () { return { status: 200, body: \\\"ok\\\" }; }\"))(__mod, __mod.exports);\nconst _fn = __mod.exports;")
src, ok := decodeNodeJSWrapperSource(wrapper)
if !ok {
t.Fatalf("decodeNodeJSWrapperSource() = false, want true")
}
want := "module.exports = async function () { return { status: 200, body: \"ok\" }; }"
if src != want {
t.Fatalf("decoded source = %q, want %q", src, want)
}
}
func TestDecodeArchiveBytesToSourcePrefersNodeJSWrapperExtraction(t *testing.T) {
wrapper := []byte("const __mod = { exports: {} };\n(new Function('module', 'exports', \"module.exports = async function () { return { status: 200, body: \\\"ok\\\" }; }\"))(__mod, __mod.exports);\nconst _fn = __mod.exports;\n\n\tasync function __invoke(ctx) {\n const fn = typeof _fn === 'function' ? _fn : (_fn.default || _fn.handler || _fn.main);\n if (!fn) throw new Error('no exported function found in user code');\n const result = await fn(ctx);\n if (!result) return { status: 200, body: '' };\n if (typeof result.status !== 'undefined') return result;\n return { status: 200, ...result };\n}\n\n\texport { __invoke as main, __invoke as handler };\n\texport default __invoke;\n")
src, err := decodeArchiveBytesToSource(wrapper)
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
want := "module.exports = async function () { return { status: 200, body: \"ok\" }; }"
if src != want {
t.Fatalf("decoded source = %q, want %q", src, want)
}
}
func TestDecodeArchiveBytesToSourcePrefersNodeJSWrapperExtractionWithNewlines(t *testing.T) {
wrapper := []byte("const __mod = { exports: {} };\n(new Function('module', 'exports', \"module.exports = async function (context) {\\n const value = await Promise.resolve(context?.value || 'ok');\\n return { status: 200, body: value };\\n}\"))(__mod, __mod.exports);\nconst _fn = __mod.exports;\n\n\tasync function __invoke(ctx) {\n const fn = typeof _fn === 'function' ? _fn : (_fn.default || _fn.handler || _fn.main);\n if (!fn) throw new Error('no exported function found in user code');\n const result = await fn(ctx);\n if (!result) return { status: 200, body: '' };\n if (typeof result.status !== 'undefined') return result;\n return { status: 200, ...result };\n}\n\n\texport { __invoke as main, __invoke as handler };\n\texport default __invoke;\n")
src, err := decodeArchiveBytesToSource(wrapper)
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
want := "module.exports = async function (context) {\n const value = await Promise.resolve(context?.value || 'ok');\n return { status: 200, body: value };\n}"
if src != want {
t.Fatalf("decoded source = %q, want %q", src, want)
}
}
func TestDecodeArchiveBytesToSourceKeepsPythonMultilineSource(t *testing.T) {
src := "def main(context):\n value = context.get('value', 'ok')\n return {'status': 200, 'body': value}\n"
got, err := decodeArchiveBytesToSource([]byte(src))
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
if got != src {
t.Fatalf("decoded source = %q, want %q", got, src)
}
}
func TestDecodeArchiveBytesToSourceUnwrapsNodeJSZipPayload(t *testing.T) {
zipBytes, err := runtime.BuildJSDeployZip(`module.exports = async function (context) {
return { status: 200, body: context?.value || "ok" };
}`)
if err != nil {
t.Fatalf("runtime.BuildJSDeployZip() error = %v", err)
}
got, err := decodeArchiveBytesToSource(zipBytes)
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
want := `module.exports = async function (context) {
return { status: 200, body: context?.value || "ok" };
}`
if got != want {
t.Fatalf("decoded source = %q, want %q", got, want)
}
}
func TestDecodeArchiveBytesToSourceKeepsRubyMultilineSource(t *testing.T) {
src := "def handler(context)\n value = context[:value] || 'ok'\n { status: 200, body: value }\nend\n"
got, err := decodeArchiveBytesToSource([]byte(src))
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
if got != src {
t.Fatalf("decoded source = %q, want %q", got, src)
}
}
func TestDecodeArchiveBytesToSourceKeepsPHPMultilineSource(t *testing.T) {
src := "<?php\nfunction handler(array $context): array {\n $value = $context['value'] ?? 'ok';\n return ['status' => 200, 'body' => $value];\n}\n"
got, err := decodeArchiveBytesToSource([]byte(src))
if err != nil {
t.Fatalf("decodeArchiveBytesToSource() error = %v", err)
}
if got != src {
t.Fatalf("decoded source = %q, want %q", got, src)
}
}