“Naeel”
f42fa366de
handlers.go (267 lines) — routing, constants, handleAuth, utils storagesvc.go — S3 upload/delete via storagesvc function_code.go — create/update function from inline code function_archive.go — create/update function from zip archive function_crud.go — GET, DELETE, timeout update, pod logs function_invoke.go — invoke via router, /fn route gateway, cron gateway No logic changes. go build OK.
424 lines
15 KiB
Go
424 lines
15 KiB
Go
// Package api — вызов функций через Fission router.
|
||
//
|
||
// Этот файл содержит три способа вызова функций:
|
||
// - handleInvokeFunction — POST /functions/:name/invoke (через консоль, для тестирования)
|
||
// - handleInvokeRoute — /fn/<route> (публичный gateway, пользователь вызывает по своему маршруту)
|
||
// - handleFissionFunctionGateway — /fission-function/<ns>/<name> (внутренний gateway для cron/timer)
|
||
//
|
||
// Все три варианта проксируют запрос к Fission router с JWT-токеном router.
|
||
// Таймаут вызова берётся из spec.functionTimeout функции (или из конфига если не задан).
|
||
//
|
||
// Вспомогательные утилиты (buildInternalInvokeURL, copyProxyRequestHeaders и др.) — в этом же файле.
|
||
package api
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"fission-console/internal/fission"
|
||
|
||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||
)
|
||
|
||
// resolveInvokeTimeout возвращает таймаут вызова функции.
|
||
// Приоритет: spec.functionTimeout функции → конфиг сервера → defaultFunctionInvokeTimeout.
|
||
func (s *Server) resolveInvokeTimeout(fn *unstructured.Unstructured) time.Duration {
|
||
if fn != nil {
|
||
seconds, found, err := unstructured.NestedInt64(fn.Object, "spec", "functionTimeout")
|
||
if err == nil && found && seconds > 0 {
|
||
return time.Duration(seconds) * time.Second
|
||
}
|
||
}
|
||
if s.invokeTimeout > 0 {
|
||
return s.invokeTimeout
|
||
}
|
||
return defaultFunctionInvokeTimeout
|
||
}
|
||
|
||
// buildInternalInvokeURL строит URL для вызова функции через Fission router.
|
||
// Для namespace "default" — /fission-function/<name>.
|
||
// Для остальных — /fission-function/<namespace>/<name>.
|
||
func buildInternalInvokeURL(routerURL, namespace, functionName string) string {
|
||
if namespace == "default" || namespace == "" {
|
||
return fmt.Sprintf("%s/fission-function/%s", routerURL, functionName)
|
||
}
|
||
return fmt.Sprintf("%s/fission-function/%s/%s", routerURL, namespace, functionName)
|
||
}
|
||
|
||
// shouldForwardRequestBody возвращает true если метод подразумевает тело запроса.
|
||
// GET и HEAD не имеют тела — тело не проксируется.
|
||
func shouldForwardRequestBody(method string) bool {
|
||
switch method {
|
||
case http.MethodGet, http.MethodHead:
|
||
return false
|
||
default:
|
||
return true
|
||
}
|
||
}
|
||
|
||
// copyProxyRequestHeaders копирует заголовки из входящего запроса в исходящий.
|
||
// Пропускаем служебные заголовки чтобы не перезаписать их при проксировании.
|
||
func copyProxyRequestHeaders(dst, src http.Header) {
|
||
for key, values := range src {
|
||
switch http.CanonicalHeaderKey(key) {
|
||
case "Authorization", "X-Auth-Token", "X-Auth-Env", "Host", "Content-Length":
|
||
continue
|
||
}
|
||
for _, value := range values {
|
||
dst.Add(key, value)
|
||
}
|
||
}
|
||
}
|
||
|
||
// copyProxyResponseHeaders копирует все заголовки из upstream-ответа в ответ клиенту.
|
||
func copyProxyResponseHeaders(dst, src http.Header) {
|
||
for key, values := range src {
|
||
for _, value := range values {
|
||
dst.Add(key, value)
|
||
}
|
||
}
|
||
}
|
||
|
||
// doRequestWithContextTimeout выполняет HTTP-запрос без глобального таймаута клиента.
|
||
// Реальный лимит задаётся через context — это позволяет функции иметь свой таймаут
|
||
// независимо от общего HTTP-таймаута console.
|
||
func doRequestWithContextTimeout(client *http.Client, req *http.Request) (*http.Response, error) {
|
||
if client == nil {
|
||
return http.DefaultClient.Do(req)
|
||
}
|
||
invokeClient := *client
|
||
invokeClient.Timeout = 0
|
||
return invokeClient.Do(req)
|
||
}
|
||
|
||
// handleInvokeFunction вызывает функцию через Fission router.
|
||
// Определяет реальный URL из HTTPTrigger, выбирает метод (POST/GET).
|
||
func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, name string) {
|
||
bodyBytes, err := io.ReadAll(r.Body)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("read request body: %v", err))
|
||
return
|
||
}
|
||
if len(bytes.TrimSpace(bodyBytes)) == 0 {
|
||
bodyBytes = []byte("{}")
|
||
}
|
||
|
||
ns := s.userNS(r)
|
||
lookupCtx, lookupCancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||
defer lookupCancel()
|
||
|
||
// Проверяем существование функции до вызова — лучше 404 чем непонятный timeout
|
||
fn, err2 := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Get(lookupCtx, name, metav1.GetOptions{})
|
||
if err2 != nil {
|
||
if apierrors.IsNotFound(err2) {
|
||
writeJSONError(w, http.StatusNotFound, fmt.Sprintf("function %q not found", name))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("get function %q: %v", name, err2))
|
||
return
|
||
}
|
||
|
||
invokeTimeout := s.resolveInvokeTimeout(fn)
|
||
ctx, cancel := context.WithTimeout(r.Context(), invokeTimeout)
|
||
defer cancel()
|
||
|
||
// Ищем HTTPTrigger чтобы получить реальный URL и метод
|
||
invokeURL := buildInternalInvokeURL(s.routerURL, ns, name)
|
||
invokeMethod := http.MethodPost
|
||
triggers, err := s.dyn.Resource(fission.HTTPTrigGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
|
||
if err == nil {
|
||
for _, trig := range triggers.Items {
|
||
refName, _, _ := unstructured.NestedString(trig.Object, "spec", "functionref", "name")
|
||
if refName != name {
|
||
continue
|
||
}
|
||
route, _, _ := unstructured.NestedString(trig.Object, "spec", "relativeurl")
|
||
methods, _, _ := unstructured.NestedStringSlice(trig.Object, "spec", "methods")
|
||
hasPost, hasGet := false, false
|
||
for _, m := range methods {
|
||
switch strings.ToUpper(strings.TrimSpace(m)) {
|
||
case http.MethodPost:
|
||
hasPost = true
|
||
case http.MethodGet:
|
||
hasGet = true
|
||
}
|
||
}
|
||
if route != "" {
|
||
if !strings.HasPrefix(route, "/") {
|
||
route = "/" + route
|
||
}
|
||
invokeURL = s.routerURL + route
|
||
// Если функция поддерживает только GET — используем GET
|
||
if !hasPost && hasGet {
|
||
invokeMethod = http.MethodGet
|
||
}
|
||
break
|
||
}
|
||
}
|
||
}
|
||
|
||
start := time.Now()
|
||
var invokeBody io.Reader
|
||
if invokeMethod == http.MethodPost {
|
||
invokeBody = bytes.NewReader(bodyBytes)
|
||
}
|
||
|
||
req, err := http.NewRequestWithContext(ctx, invokeMethod, invokeURL, invokeBody)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("build invoke request: %v", err))
|
||
return
|
||
}
|
||
if invokeMethod == http.MethodPost {
|
||
req.Header.Set("Content-Type", "application/json")
|
||
}
|
||
if token := s.getRouterToken(); token != "" {
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
}
|
||
|
||
resp, err := doRequestWithContextTimeout(s.http, req)
|
||
if err != nil {
|
||
// Отличаем timeout от сетевой ошибки.
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q timeout after %s", name, invokeTimeout))
|
||
return
|
||
}
|
||
var netErr net.Error
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q timeout after %s", name, invokeTimeout))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q: %v", name, err))
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
respBody, _ := io.ReadAll(resp.Body)
|
||
writeAnyJSON(w, http.StatusOK, map[string]any{
|
||
"status": resp.StatusCode,
|
||
"latency_ms": time.Since(start).Milliseconds(),
|
||
"response_raw": string(respBody),
|
||
})
|
||
}
|
||
|
||
// handleFissionFunctionGateway принимает внутренние invoke-запросы timer/router
|
||
// и проксирует их через console в upstream router с корректным router JWT.
|
||
func (s *Server) handleFissionFunctionGateway(w http.ResponseWriter, r *http.Request) {
|
||
rawPath := strings.Trim(strings.TrimPrefix(r.URL.Path, "/fission-function"), "/")
|
||
if rawPath == "" {
|
||
http.NotFound(w, r)
|
||
return
|
||
}
|
||
|
||
parts := strings.Split(rawPath, "/")
|
||
namespace := s.ns
|
||
functionName := ""
|
||
remainingPath := ""
|
||
|
||
if len(parts) == 1 {
|
||
functionName = strings.TrimSpace(parts[0])
|
||
} else {
|
||
namespace = strings.TrimSpace(parts[0])
|
||
functionName = strings.TrimSpace(parts[1])
|
||
if len(parts) > 2 {
|
||
remainingPath = "/" + strings.Join(parts[2:], "/")
|
||
}
|
||
}
|
||
|
||
if namespace == "" || functionName == "" {
|
||
writeJSONError(w, http.StatusBadRequest, "namespace and function name are required")
|
||
return
|
||
}
|
||
|
||
s.invokeInternalFunction(w, r, namespace, functionName, remainingPath)
|
||
}
|
||
|
||
// invokeInternalFunction проксирует вызов функции к Fission router.
|
||
// Используется как из handleFissionFunctionGateway (cron/timer), так и из handleInvokeRoute.
|
||
func (s *Server) invokeInternalFunction(w http.ResponseWriter, r *http.Request, namespace, functionName, extraPath string) {
|
||
bodyBytes, err := io.ReadAll(r.Body)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("read request body: %v", err))
|
||
return
|
||
}
|
||
if len(bytes.TrimSpace(bodyBytes)) == 0 {
|
||
bodyBytes = []byte("{}")
|
||
}
|
||
|
||
lookupCtx, lookupCancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||
defer lookupCancel()
|
||
|
||
fn, err := s.dyn.Resource(fission.FunctionGVR).Namespace(namespace).Get(lookupCtx, functionName, metav1.GetOptions{})
|
||
if err != nil {
|
||
if apierrors.IsNotFound(err) {
|
||
writeJSONError(w, http.StatusNotFound, fmt.Sprintf("function %q not found", functionName))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("get function %q: %v", functionName, err))
|
||
return
|
||
}
|
||
|
||
invokeTimeout := s.resolveInvokeTimeout(fn)
|
||
ctx, cancel := context.WithTimeout(r.Context(), invokeTimeout)
|
||
defer cancel()
|
||
|
||
invokeURL := buildInternalInvokeURL(s.routerURL, namespace, functionName) + extraPath
|
||
if r.URL.RawQuery != "" {
|
||
invokeURL += "?" + r.URL.RawQuery
|
||
}
|
||
|
||
var invokeBody io.Reader
|
||
if shouldForwardRequestBody(r.Method) {
|
||
invokeBody = bytes.NewReader(bodyBytes)
|
||
}
|
||
|
||
req, err := http.NewRequestWithContext(ctx, r.Method, invokeURL, invokeBody)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("build invoke request: %v", err))
|
||
return
|
||
}
|
||
if shouldForwardRequestBody(r.Method) {
|
||
req.Header.Set("Content-Type", "application/json")
|
||
}
|
||
copyProxyRequestHeaders(req.Header, r.Header)
|
||
if token := s.getRouterToken(); token != "" {
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
}
|
||
|
||
start := time.Now()
|
||
resp, err := doRequestWithContextTimeout(s.http, req)
|
||
if err != nil {
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q timeout after %s", functionName, invokeTimeout))
|
||
return
|
||
}
|
||
var netErr net.Error
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q timeout after %s", functionName, invokeTimeout))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q: %v", functionName, err))
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
respBody, _ := io.ReadAll(resp.Body)
|
||
writeAnyJSON(w, http.StatusOK, map[string]any{
|
||
"status": resp.StatusCode,
|
||
"latency_ms": time.Since(start).Milliseconds(),
|
||
"response_raw": string(respBody),
|
||
})
|
||
}
|
||
|
||
// handleInvokeRoute даёт пользователю прямой HTTP gateway к своей функции по route.
|
||
// Внешний контракт: /fn/<route> + Authorization: Bearer <user-token>.
|
||
func (s *Server) handleInvokeRoute(w http.ResponseWriter, r *http.Request) {
|
||
route := normalizeRoute(strings.TrimPrefix(r.URL.Path, "/fn"))
|
||
if route == "/" {
|
||
writeJSONError(w, http.StatusBadRequest, "route is required")
|
||
return
|
||
}
|
||
|
||
ns := s.userNS(r)
|
||
lookupCtx, lookupCancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||
defer lookupCancel()
|
||
|
||
triggers, err := s.dyn.Resource(fission.HTTPTrigGVR).Namespace(ns).List(lookupCtx, metav1.ListOptions{})
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("list httptriggers: %v", err))
|
||
return
|
||
}
|
||
|
||
matchedFunction := ""
|
||
allowedMethods := make([]string, 0, 4)
|
||
for _, trig := range triggers.Items {
|
||
trigRoute, _, _ := unstructured.NestedString(trig.Object, "spec", "relativeurl")
|
||
if normalizeRoute(trigRoute) != route {
|
||
continue
|
||
}
|
||
methods, _, _ := unstructured.NestedStringSlice(trig.Object, "spec", "methods")
|
||
allowedMethods = appendUniqueMethods(allowedMethods, methods)
|
||
if !routeAllowsMethod(methods, r.Method) {
|
||
continue
|
||
}
|
||
matchedFunction, _, _ = unstructured.NestedString(trig.Object, "spec", "functionref", "name")
|
||
if matchedFunction != "" {
|
||
break
|
||
}
|
||
}
|
||
|
||
if matchedFunction == "" {
|
||
if len(allowedMethods) > 0 {
|
||
w.Header().Set("Allow", strings.Join(allowedMethods, ", "))
|
||
writeJSONError(w, http.StatusMethodNotAllowed, fmt.Sprintf("route %q does not allow method %s", route, r.Method))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusNotFound, fmt.Sprintf("route %q not found", route))
|
||
return
|
||
}
|
||
|
||
fn, err := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Get(lookupCtx, matchedFunction, metav1.GetOptions{})
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("get function %q: %v", matchedFunction, err))
|
||
return
|
||
}
|
||
|
||
bodyBytes, err := io.ReadAll(r.Body)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("read request body: %v", err))
|
||
return
|
||
}
|
||
|
||
invokeTimeout := s.resolveInvokeTimeout(fn)
|
||
ctx, cancel := context.WithTimeout(r.Context(), invokeTimeout)
|
||
defer cancel()
|
||
|
||
invokeURL := s.routerURL + route
|
||
if r.URL.RawQuery != "" {
|
||
invokeURL += "?" + r.URL.RawQuery
|
||
}
|
||
|
||
var invokeBody io.Reader
|
||
if shouldForwardRequestBody(r.Method) {
|
||
invokeBody = bytes.NewReader(bodyBytes)
|
||
}
|
||
|
||
req, err := http.NewRequestWithContext(ctx, r.Method, invokeURL, invokeBody)
|
||
if err != nil {
|
||
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("build invoke request: %v", err))
|
||
return
|
||
}
|
||
copyProxyRequestHeaders(req.Header, r.Header)
|
||
if token := s.getRouterToken(); token != "" {
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
}
|
||
|
||
resp, err := doRequestWithContextTimeout(s.http, req)
|
||
if err != nil {
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke route %q timeout after %s", route, invokeTimeout))
|
||
return
|
||
}
|
||
var netErr net.Error
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke route %q timeout after %s", route, invokeTimeout))
|
||
return
|
||
}
|
||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke route %q: %v", route, err))
|
||
return
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
copyProxyResponseHeaders(w.Header(), resp.Header)
|
||
w.WriteHeader(resp.StatusCode)
|
||
_, _ = io.Copy(w, resp.Body)
|
||
}
|