console: use per-function invoke timeout
This commit is contained in:
@@ -34,6 +34,9 @@ var validFuncName = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?$`)
|
||||
// Выше — не имеет смысла для inline функции; лучше использовать Package с URL.
|
||||
const maxCodeSize = 1 << 20
|
||||
|
||||
// defaultFunctionInvokeTimeout совпадает с дефолтом Fission для spec.functionTimeout.
|
||||
const defaultFunctionInvokeTimeout = 60 * time.Second
|
||||
|
||||
func buildDeployArchive(lang, code string) ([]byte, error) {
|
||||
switch lang {
|
||||
case "nodejs":
|
||||
@@ -47,6 +50,26 @@ func buildDeployArchive(lang, code string) ([]byte, error) {
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
func normalizeFunctionTimeout(seconds int64) int64 {
|
||||
if seconds <= 0 {
|
||||
return int64(defaultFunctionInvokeTimeout / time.Second)
|
||||
}
|
||||
return seconds
|
||||
}
|
||||
|
||||
// handleFunctionsRoot обрабатывает запросы к /console/api/functions без имени функции.
|
||||
// GET → список всех функций, POST → создать новую.
|
||||
func (s *Server) handleFunctionsRoot(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -184,6 +207,7 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
|
||||
req.Route = "/" + req.Route
|
||||
}
|
||||
req.Methods = normalizeMethods(req.Methods)
|
||||
req.Timeout = normalizeFunctionTimeout(req.Timeout)
|
||||
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
|
||||
defer cancel()
|
||||
@@ -275,6 +299,7 @@ func (s *Server) handleCreateFunction(w http.ResponseWriter, r *http.Request) {
|
||||
"metadata": map[string]any{"name": req.Name, "namespace": ns, "annotations": fnAnnotations},
|
||||
"spec": map[string]any{
|
||||
"environment": map[string]any{"name": req.Environment, "namespace": ns},
|
||||
"functionTimeout": req.Timeout,
|
||||
"InvokeStrategy": map[string]any{
|
||||
"ExecutionStrategy": map[string]any{"ExecutorType": "poolmgr"},
|
||||
"StrategyType": "execution",
|
||||
@@ -345,6 +370,10 @@ func (s *Server) handleGetFunction(w http.ResponseWriter, r *http.Request, name
|
||||
packageName, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "packageref", "name")
|
||||
environment, _, _ := unstructured.NestedString(fn.Object, "spec", "environment", "name")
|
||||
entrypoint, _, _ := unstructured.NestedString(fn.Object, "spec", "package", "functionName")
|
||||
functionTimeout, foundTimeout, _ := unstructured.NestedInt64(fn.Object, "spec", "functionTimeout")
|
||||
if !foundTimeout || functionTimeout <= 0 {
|
||||
functionTimeout = int64(defaultFunctionInvokeTimeout / time.Second)
|
||||
}
|
||||
|
||||
// Извлекаем исходный код из Package (пробуем source.literal, потом deployment.literal, потом url)
|
||||
code := ""
|
||||
@@ -377,6 +406,7 @@ func (s *Server) handleGetFunction(w http.ResponseWriter, r *http.Request, name
|
||||
"environment": environment,
|
||||
"package": packageName,
|
||||
"entrypoint": entrypoint,
|
||||
"timeout": functionTimeout,
|
||||
"code": code,
|
||||
"route": route,
|
||||
"methods": methods,
|
||||
@@ -452,6 +482,11 @@ func (s *Server) handleUpdateFunctionCode(w http.ResponseWriter, r *http.Request
|
||||
return
|
||||
}
|
||||
|
||||
if err := unstructured.SetNestedField(fn.Object, normalizeFunctionTimeout(req.Timeout), "spec", "functionTimeout"); err != nil {
|
||||
writeJSONError(w, http.StatusInternalServerError, fmt.Sprintf("set function timeout: %v", err))
|
||||
return
|
||||
}
|
||||
|
||||
// Синхронизируем resourceVersion в Function.spec.package.packageref
|
||||
// Это триггерит executor перезагрузить код в pool pod
|
||||
if err := unstructured.SetNestedField(fn.Object, updatedPkg.GetResourceVersion(), "spec", "package", "packageref", "resourceversion"); err != nil {
|
||||
@@ -482,17 +517,13 @@ func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, na
|
||||
bodyBytes = []byte("{}")
|
||||
}
|
||||
|
||||
invokeTimeout := s.invokeTimeout
|
||||
if invokeTimeout <= 0 {
|
||||
invokeTimeout = 20 * time.Second
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(r.Context(), invokeTimeout)
|
||||
defer cancel()
|
||||
ns := s.userNS(r)
|
||||
lookupCtx, lookupCancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||||
defer lookupCancel()
|
||||
|
||||
// Проверяем существование функции до вызова — лучше 404 чем непонятный timeout
|
||||
if _, err2 := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Get(ctx, name, metav1.GetOptions{}); err2 != nil {
|
||||
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
|
||||
@@ -501,6 +532,10 @@ func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, na
|
||||
return
|
||||
}
|
||||
|
||||
invokeTimeout := s.resolveInvokeTimeout(fn)
|
||||
ctx, cancel := context.WithTimeout(r.Context(), invokeTimeout)
|
||||
defer cancel()
|
||||
|
||||
// Ищем HTTPTrigger чтобы получить реальный URL и метод
|
||||
invokeURL := fmt.Sprintf("%s/fission-function/v2/functions/%s", s.routerURL, name)
|
||||
invokeMethod := http.MethodPost
|
||||
@@ -556,14 +591,14 @@ func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, na
|
||||
|
||||
resp, err := s.http.Do(req)
|
||||
if err != nil {
|
||||
// Отличаем timeout от сетевой ошибки — timeout часто означает что функция не специализировалась
|
||||
// Отличаем timeout от сетевой ошибки.
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("invoke %q timeout after %s: function specialization likely failed (for example, syntax error)", name, invokeTimeout))
|
||||
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: function specialization likely failed (for example, syntax error)", name, invokeTimeout))
|
||||
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))
|
||||
|
||||
Reference in New Issue
Block a user