fix: document cron router auth chain
This commit is contained in:
Naeel
@@ -0,0 +1,203 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type CronMetric struct {
|
||||
Timestamp string `json:"timestamp"`
|
||||
Source string `json:"source"`
|
||||
Hostname string `json:"hostname,omitempty"`
|
||||
Status string `json:"status,omitempty"`
|
||||
MemoryTotalMB float64 `json:"memory_total_mb,omitempty"`
|
||||
MemoryFreeMB float64 `json:"memory_free_mb,omitempty"`
|
||||
MemoryAvailMB float64 `json:"memory_available_mb,omitempty"`
|
||||
MemoryUsedMB float64 `json:"memory_used_mb,omitempty"`
|
||||
MemoryPercent float64 `json:"memory_percent,omitempty"`
|
||||
CpuLoad1m float64 `json:"cpu_load_1m,omitempty"`
|
||||
CpuLoad5m float64 `json:"cpu_load_5m,omitempty"`
|
||||
CpuLoad15m float64 `json:"cpu_load_15m,omitempty"`
|
||||
DiskTotalGB float64 `json:"disk_total_gb,omitempty"`
|
||||
DiskFreeGB float64 `json:"disk_free_gb,omitempty"`
|
||||
DiskUsedGB float64 `json:"disk_used_gb,omitempty"`
|
||||
UptimeSec float64 `json:"uptime_sec,omitempty"`
|
||||
Notes string `json:"notes,omitempty"`
|
||||
}
|
||||
|
||||
var (
|
||||
cronMetricMu sync.RWMutex
|
||||
cronMetricHistory []CronMetric
|
||||
cronMetricMaxKeep = 240
|
||||
)
|
||||
|
||||
func (s *Server) handleCronMetrics(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.Method {
|
||||
case http.MethodGet:
|
||||
cronMetricMu.RLock()
|
||||
history := append([]CronMetric(nil), cronMetricHistory...)
|
||||
cronMetricMu.RUnlock()
|
||||
latest := CronMetric{}
|
||||
if len(history) > 0 {
|
||||
latest = history[len(history)-1]
|
||||
}
|
||||
writeAnyJSON(w, http.StatusOK, map[string]any{
|
||||
"count": len(history),
|
||||
"latest": latest,
|
||||
"history": history,
|
||||
})
|
||||
case http.MethodPost:
|
||||
if !allowCronIngest(r) {
|
||||
writeJSONError(w, http.StatusUnauthorized, "invalid cron token")
|
||||
return
|
||||
}
|
||||
var payload map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
|
||||
writeJSONError(w, http.StatusBadRequest, "invalid json payload")
|
||||
return
|
||||
}
|
||||
sample := normalizeCronMetric(payload)
|
||||
cronMetricMu.Lock()
|
||||
cronMetricHistory = append(cronMetricHistory, sample)
|
||||
if len(cronMetricHistory) > cronMetricMaxKeep {
|
||||
cronMetricHistory = append([]CronMetric(nil), cronMetricHistory[len(cronMetricHistory)-cronMetricMaxKeep:]...)
|
||||
}
|
||||
count := len(cronMetricHistory)
|
||||
cronMetricMu.Unlock()
|
||||
writeAnyJSON(w, http.StatusOK, map[string]any{"ok": true, "stored": count, "latest": sample})
|
||||
default:
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
}
|
||||
|
||||
func allowCronIngest(r *http.Request) bool {
|
||||
expected := strings.TrimSpace(os.Getenv("CRON_TOKEN"))
|
||||
if expected == "" {
|
||||
return true
|
||||
}
|
||||
return strings.TrimSpace(r.Header.Get("X-Cron-Token")) == expected
|
||||
}
|
||||
|
||||
func normalizeCronMetric(payload map[string]any) CronMetric {
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
metric := CronMetric{
|
||||
Timestamp: now,
|
||||
Source: textValue(payload, "source", "function", "name", "host", "hostname", "instance"),
|
||||
Hostname: textValue(payload, "hostname", "host", "node", "instance"),
|
||||
Status: textValue(payload, "status"),
|
||||
Notes: textValue(payload, "notes", "note", "message"),
|
||||
}
|
||||
if metric.Source == "" {
|
||||
metric.Source = "cron"
|
||||
}
|
||||
|
||||
metric.MemoryTotalMB = numberValue(payload, "memory_total_mb", "total_memory_mb", "mem_total_mb", "mem_total")
|
||||
metric.MemoryFreeMB = numberValue(payload, "memory_free_mb", "free_memory_mb", "mem_free_mb", "free_mb")
|
||||
metric.MemoryAvailMB = numberValue(payload, "memory_available_mb", "available_memory_mb", "mem_available_mb", "avail_memory_mb")
|
||||
metric.MemoryUsedMB = numberValue(payload, "memory_used_mb", "used_memory_mb", "mem_used_mb", "used_mb")
|
||||
metric.MemoryPercent = numberValue(payload, "memory_percent", "mem_percent", "memory_usage_percent")
|
||||
if metric.MemoryUsedMB == 0 && metric.MemoryTotalMB > 0 && metric.MemoryFreeMB > 0 {
|
||||
metric.MemoryUsedMB = metric.MemoryTotalMB - metric.MemoryFreeMB
|
||||
}
|
||||
if metric.MemoryAvailMB == 0 && metric.MemoryFreeMB > 0 {
|
||||
metric.MemoryAvailMB = metric.MemoryFreeMB
|
||||
}
|
||||
if metric.MemoryPercent == 0 && metric.MemoryTotalMB > 0 && metric.MemoryUsedMB > 0 {
|
||||
metric.MemoryPercent = (metric.MemoryUsedMB / metric.MemoryTotalMB) * 100
|
||||
}
|
||||
|
||||
metric.CpuLoad1m = numberValue(payload, "cpu_load_1m", "loadavg_1m", "load_1m", "load1")
|
||||
metric.CpuLoad5m = numberValue(payload, "cpu_load_5m", "loadavg_5m", "load_5m", "load5")
|
||||
metric.CpuLoad15m = numberValue(payload, "cpu_load_15m", "loadavg_15m", "load_15m", "load15")
|
||||
|
||||
metric.DiskTotalGB = numberValue(payload, "disk_total_gb", "total_disk_gb", "disk_total_mb", "disk_total")
|
||||
metric.DiskFreeGB = numberValue(payload, "disk_free_gb", "free_disk_gb", "disk_free_mb", "disk_free")
|
||||
metric.DiskUsedGB = numberValue(payload, "disk_used_gb", "used_disk_gb", "disk_used_mb", "disk_used")
|
||||
if metric.DiskUsedGB == 0 && metric.DiskTotalGB > 0 && metric.DiskFreeGB > 0 {
|
||||
metric.DiskUsedGB = metric.DiskTotalGB - metric.DiskFreeGB
|
||||
}
|
||||
|
||||
if uptime := numberValue(payload, "uptime_sec", "uptime", "uptime_seconds"); uptime > 0 {
|
||||
metric.UptimeSec = uptime
|
||||
}
|
||||
if ts := textValue(payload, "timestamp", "ts", "collected_at"); ts != "" {
|
||||
metric.Timestamp = ts
|
||||
}
|
||||
return metric
|
||||
}
|
||||
|
||||
func textValue(payload map[string]any, keys ...string) string {
|
||||
for _, key := range keys {
|
||||
if value, ok := payload[key]; ok {
|
||||
text := strings.TrimSpace(toString(value))
|
||||
if text != "" {
|
||||
return text
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func numberValue(payload map[string]any, keys ...string) float64 {
|
||||
for _, key := range keys {
|
||||
if value, ok := payload[key]; ok {
|
||||
if number, ok := toFloat64(value); ok {
|
||||
return number
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func toString(value any) string {
|
||||
switch v := value.(type) {
|
||||
case string:
|
||||
return v
|
||||
case []byte:
|
||||
return string(v)
|
||||
case float64:
|
||||
return strconv.FormatFloat(v, 'f', -1, 64)
|
||||
case float32:
|
||||
return strconv.FormatFloat(float64(v), 'f', -1, 32)
|
||||
case int:
|
||||
return strconv.FormatInt(int64(v), 10)
|
||||
case int64:
|
||||
return strconv.FormatInt(v, 10)
|
||||
case json.Number:
|
||||
return v.String()
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
func toFloat64(value any) (float64, bool) {
|
||||
switch v := value.(type) {
|
||||
case float64:
|
||||
return v, true
|
||||
case float32:
|
||||
return float64(v), true
|
||||
case int:
|
||||
return float64(v), true
|
||||
case int64:
|
||||
return float64(v), true
|
||||
case json.Number:
|
||||
f, err := v.Float64()
|
||||
return f, err == nil
|
||||
case string:
|
||||
if strings.TrimSpace(v) == "" {
|
||||
return 0, false
|
||||
}
|
||||
n, err := strconv.ParseFloat(strings.TrimSpace(v), 64)
|
||||
if err == nil {
|
||||
return n, true
|
||||
}
|
||||
return 0, false
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
@@ -654,7 +654,7 @@ func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, na
|
||||
defer cancel()
|
||||
|
||||
// Ищем HTTPTrigger чтобы получить реальный URL и метод
|
||||
invokeURL := fmt.Sprintf("%s/fission-function/v2/functions/%s", s.routerURL, name)
|
||||
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 {
|
||||
@@ -732,6 +732,13 @@ func (s *Server) handleInvokeFunction(w http.ResponseWriter, r *http.Request, na
|
||||
})
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
// handleInvokeRoute даёт пользователю прямой HTTP gateway к своей функции по route.
|
||||
// Внешний контракт: /fn/<route> + Authorization: Bearer <user-token>.
|
||||
func (s *Server) handleInvokeRoute(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -836,7 +843,7 @@ func (s *Server) handleInvokeRoute(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = io.Copy(w, resp.Body)
|
||||
}
|
||||
|
||||
// handleDeleteFunction удаляет функцию и связанные объекты: HTTPTrigger, Package.
|
||||
// handleDeleteFunction удаляет функцию и связанные объекты: HTTPTrigger, TimeTrigger, Package.
|
||||
// После удаления вызывает CleanupEnvironmentIfUnused — убирает environment если язык больше не используется.
|
||||
func (s *Server) handleDeleteFunction(w http.ResponseWriter, r *http.Request, name string) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
|
||||
@@ -868,6 +875,16 @@ func (s *Server) handleDeleteFunction(w http.ResponseWriter, r *http.Request, na
|
||||
}
|
||||
}
|
||||
|
||||
// Удаляем связанные TimeTrigger-ы
|
||||
if triggers, err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).List(ctx, metav1.ListOptions{}); err == nil {
|
||||
for _, trig := range triggers.Items {
|
||||
refName, _, _ := unstructured.NestedString(trig.Object, "spec", "functionref", "name")
|
||||
if refName == name {
|
||||
_ = s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).Delete(ctx, trig.GetName(), metav1.DeleteOptions{})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
|
||||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("delete function %q: %v", name, err))
|
||||
return
|
||||
|
||||
@@ -124,6 +124,11 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) {
|
||||
uiHandler := http.StripPrefix("/console", ui.Handler())
|
||||
mux.Handle("/console", uiHandler)
|
||||
mux.Handle("/console/", uiHandler)
|
||||
cronHandler := ui.CronHandler()
|
||||
mux.Handle("/cron", cronHandler)
|
||||
mux.Handle("/cron/", cronHandler)
|
||||
mux.Handle("/console/cron", cronHandler)
|
||||
mux.Handle("/console/cron/", cronHandler)
|
||||
|
||||
// Auth: не требует токена — сам проверяет и возвращает namespace
|
||||
mux.HandleFunc("/console/api/auth", s.handleAuth)
|
||||
@@ -137,7 +142,10 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/api/functions", s.handleFunctionsRoot)
|
||||
mux.HandleFunc("/api/functions/", s.handleFunctionsAction)
|
||||
mux.HandleFunc("/api/httptriggers", s.handleList(fission.HTTPTrigGVR))
|
||||
mux.HandleFunc("/api/timetriggers", s.handleList(fission.TimeTrigGVR))
|
||||
mux.HandleFunc("/api/timetriggers", s.handleTimeTriggersRoot)
|
||||
mux.HandleFunc("/api/timetriggers/", s.handleTimeTriggersAction)
|
||||
mux.HandleFunc("/cron/api/metrics", s.handleCronMetrics)
|
||||
mux.HandleFunc("/console/cron/api/metrics", s.handleCronMetrics)
|
||||
|
||||
// Основные /console/api/* маршруты
|
||||
mux.HandleFunc("/console/api/environments", auth(s.handleList(fission.EnvironmentGVR)))
|
||||
@@ -146,7 +154,8 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/console/api/functions/", auth(s.handleFunctionsAction))
|
||||
mux.HandleFunc("/fn/", auth(s.handleInvokeRoute))
|
||||
mux.HandleFunc("/console/api/httptriggers", auth(s.handleList(fission.HTTPTrigGVR)))
|
||||
mux.HandleFunc("/console/api/timetriggers", auth(s.handleList(fission.TimeTrigGVR)))
|
||||
mux.HandleFunc("/console/api/timetriggers", auth(s.handleTimeTriggersRoot))
|
||||
mux.HandleFunc("/console/api/timetriggers/", auth(s.handleTimeTriggersAction))
|
||||
mux.HandleFunc("/console/api/ns/status", auth(s.handleNSStatus))
|
||||
mux.HandleFunc("/console/api/ns/debug", auth(s.handleNSDebug))
|
||||
mux.HandleFunc("/console/api/ai/check", auth(s.handleAICheck))
|
||||
|
||||
@@ -0,0 +1,480 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"fission-console/internal/fission"
|
||||
"fission-console/internal/model"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
)
|
||||
|
||||
const (
|
||||
timeTriggerDefaultMethod = http.MethodPost
|
||||
timeTriggerDefaultSubPath = "/"
|
||||
timeTriggerMaxNameLen = 63
|
||||
)
|
||||
|
||||
var validTimeTriggerMethods = map[string]struct{}{
|
||||
http.MethodGet: {},
|
||||
http.MethodPost: {},
|
||||
http.MethodPut: {},
|
||||
http.MethodDelete: {},
|
||||
http.MethodHead: {},
|
||||
}
|
||||
|
||||
func (s *Server) handleTimeTriggersRoot(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.Method {
|
||||
case http.MethodGet:
|
||||
s.handleList(fission.TimeTrigGVR)(w, r)
|
||||
case http.MethodPost:
|
||||
s.handleCreateTimeTrigger(w, r)
|
||||
default:
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleTimeTriggersAction(w http.ResponseWriter, r *http.Request) {
|
||||
path := strings.TrimPrefix(r.URL.Path, "/api/timetriggers/")
|
||||
if path == r.URL.Path {
|
||||
path = strings.TrimPrefix(r.URL.Path, "/console/api/timetriggers/")
|
||||
}
|
||||
path = strings.Trim(path, "/")
|
||||
if path == "" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
parts := strings.Split(path, "/")
|
||||
name := strings.TrimSpace(parts[0])
|
||||
if name == "" || len(parts) != 1 {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
switch r.Method {
|
||||
case http.MethodGet:
|
||||
s.handleGetTimeTrigger(w, r, name)
|
||||
case http.MethodPut:
|
||||
s.handleUpdateTimeTrigger(w, r, name)
|
||||
case http.MethodDelete:
|
||||
s.handleDeleteTimeTrigger(w, r, name)
|
||||
default:
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) handleCreateTimeTrigger(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.CreateTimeTriggerRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("decode request: %v", err))
|
||||
return
|
||||
}
|
||||
|
||||
ns := s.userNS(r)
|
||||
if s.nsManager != nil {
|
||||
nsCtx, nsCancel := context.WithTimeout(r.Context(), 30*time.Second)
|
||||
defer nsCancel()
|
||||
if err := s.nsManager.EnsureUserNS(nsCtx, ns); err != nil {
|
||||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("ensure namespace: %v", err))
|
||||
return
|
||||
}
|
||||
}
|
||||
if err := s.ensureTimerWatchesNamespace(r.Context(), ns); err != nil {
|
||||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("sync timer namespaces: %v", err))
|
||||
return
|
||||
}
|
||||
|
||||
trigger, err := s.createTimeTrigger(r.Context(), ns, req)
|
||||
if err != nil {
|
||||
writeJSONError(w, errStatus(err), err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
writeAnyJSON(w, http.StatusCreated, timeTriggerResponse(trigger))
|
||||
}
|
||||
|
||||
func (s *Server) handleGetTimeTrigger(w http.ResponseWriter, r *http.Request, name string) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
trigger, err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(s.userNS(r)).Get(ctx, name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
status := http.StatusBadGateway
|
||||
if apierrors.IsNotFound(err) {
|
||||
status = http.StatusNotFound
|
||||
}
|
||||
writeJSONError(w, status, fmt.Sprintf("get timetrigger %q: %v", name, err))
|
||||
return
|
||||
}
|
||||
|
||||
writeAnyJSON(w, http.StatusOK, timeTriggerResponse(trigger))
|
||||
}
|
||||
|
||||
func (s *Server) handleUpdateTimeTrigger(w http.ResponseWriter, r *http.Request, name string) {
|
||||
var req model.CreateTimeTriggerRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSONError(w, http.StatusBadRequest, fmt.Sprintf("decode request: %v", err))
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := s.ensureTimerWatchesNamespace(ctx, s.userNS(r)); err != nil {
|
||||
writeJSONError(w, http.StatusBadGateway, fmt.Sprintf("sync timer namespaces: %v", err))
|
||||
return
|
||||
}
|
||||
|
||||
trigger, err := s.updateTimeTrigger(ctx, s.userNS(r), name, req)
|
||||
if err != nil {
|
||||
writeJSONError(w, errStatus(err), err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
writeAnyJSON(w, http.StatusOK, map[string]any{
|
||||
"updated": true,
|
||||
"trigger": timeTriggerResponse(trigger),
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleDeleteTimeTrigger(w http.ResponseWriter, r *http.Request, name string) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := s.deleteTimeTrigger(ctx, s.userNS(r), name); err != nil {
|
||||
writeJSONError(w, errStatus(err), err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
writeAnyJSON(w, http.StatusOK, map[string]any{"deleted": true, "name": name})
|
||||
}
|
||||
|
||||
func (s *Server) createTimeTrigger(ctx context.Context, ns string, req model.CreateTimeTriggerRequest) (*unstructured.Unstructured, error) {
|
||||
normalized, err := normalizeTimeTriggerRequest(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if _, err := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Get(ctx, normalized.FunctionName, metav1.GetOptions{}); err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
return nil, &apiErr{status: http.StatusBadRequest, message: fmt.Sprintf("function %q not found", normalized.FunctionName)}
|
||||
}
|
||||
return nil, &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("get function %q: %v", normalized.FunctionName, err)}
|
||||
}
|
||||
|
||||
trigger := &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "fission.io/v1",
|
||||
"kind": "TimeTrigger",
|
||||
"metadata": map[string]any{
|
||||
"name": normalized.Name,
|
||||
"namespace": ns,
|
||||
},
|
||||
"spec": map[string]any{
|
||||
"cron": normalized.Cron,
|
||||
"functionref": map[string]any{
|
||||
"type": "name",
|
||||
"name": normalized.FunctionName,
|
||||
},
|
||||
"method": normalized.Method,
|
||||
"subpath": normalized.SubPath,
|
||||
},
|
||||
}}
|
||||
|
||||
created, err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).Create(ctx, trigger, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
if apierrors.IsAlreadyExists(err) {
|
||||
return nil, &apiErr{status: http.StatusConflict, message: fmt.Sprintf("timetrigger %q already exists", normalized.Name)}
|
||||
}
|
||||
if apierrors.IsInvalid(err) {
|
||||
return nil, &apiErr{status: http.StatusBadRequest, message: fmt.Sprintf("invalid timetrigger spec: %v", err)}
|
||||
}
|
||||
return nil, &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("create timetrigger: %v", err)}
|
||||
}
|
||||
|
||||
return created, nil
|
||||
}
|
||||
|
||||
func (s *Server) updateTimeTrigger(ctx context.Context, ns, name string, req model.CreateTimeTriggerRequest) (*unstructured.Unstructured, error) {
|
||||
normalized, err := normalizeTimeTriggerRequest(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
trigger, err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).Get(ctx, name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
return nil, &apiErr{status: http.StatusNotFound, message: fmt.Sprintf("timetrigger %q not found", name)}
|
||||
}
|
||||
return nil, &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("get timetrigger %q: %v", name, err)}
|
||||
}
|
||||
|
||||
if _, err := s.dyn.Resource(fission.FunctionGVR).Namespace(ns).Get(ctx, normalized.FunctionName, metav1.GetOptions{}); err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
return nil, &apiErr{status: http.StatusBadRequest, message: fmt.Sprintf("function %q not found", normalized.FunctionName)}
|
||||
}
|
||||
return nil, &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("get function %q: %v", normalized.FunctionName, err)}
|
||||
}
|
||||
|
||||
if err := unstructured.SetNestedField(trigger.Object, normalized.Cron, "spec", "cron"); err != nil {
|
||||
return nil, &apiErr{status: http.StatusInternalServerError, message: fmt.Sprintf("set timetrigger cron: %v", err)}
|
||||
}
|
||||
if err := unstructured.SetNestedField(trigger.Object, map[string]any{
|
||||
"type": "name",
|
||||
"name": normalized.FunctionName,
|
||||
}, "spec", "functionref"); err != nil {
|
||||
return nil, &apiErr{status: http.StatusInternalServerError, message: fmt.Sprintf("set timetrigger functionref: %v", err)}
|
||||
}
|
||||
if err := unstructured.SetNestedField(trigger.Object, normalized.Method, "spec", "method"); err != nil {
|
||||
return nil, &apiErr{status: http.StatusInternalServerError, message: fmt.Sprintf("set timetrigger method: %v", err)}
|
||||
}
|
||||
if err := unstructured.SetNestedField(trigger.Object, normalized.SubPath, "spec", "subpath"); err != nil {
|
||||
return nil, &apiErr{status: http.StatusInternalServerError, message: fmt.Sprintf("set timetrigger subpath: %v", err)}
|
||||
}
|
||||
|
||||
updated, err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).Update(ctx, trigger, metav1.UpdateOptions{})
|
||||
if err != nil {
|
||||
if apierrors.IsInvalid(err) {
|
||||
return nil, &apiErr{status: http.StatusBadRequest, message: fmt.Sprintf("invalid timetrigger spec: %v", err)}
|
||||
}
|
||||
return nil, &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("update timetrigger %q: %v", name, err)}
|
||||
}
|
||||
|
||||
return updated, nil
|
||||
}
|
||||
|
||||
func (s *Server) deleteTimeTrigger(ctx context.Context, ns, name string) error {
|
||||
if err := s.dyn.Resource(fission.TimeTrigGVR).Namespace(ns).Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
|
||||
return &apiErr{status: http.StatusBadGateway, message: fmt.Sprintf("delete timetrigger %q: %v", name, err)}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func normalizeTimeTriggerRequest(req model.CreateTimeTriggerRequest) (model.CreateTimeTriggerRequest, error) {
|
||||
req.Name = strings.TrimSpace(req.Name)
|
||||
req.FunctionName = strings.TrimSpace(req.FunctionName)
|
||||
req.Cron = strings.TrimSpace(req.Cron)
|
||||
req.Method = strings.TrimSpace(req.Method)
|
||||
req.SubPath = strings.TrimSpace(req.SubPath)
|
||||
|
||||
if req.Name == "" {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "name is required"}
|
||||
}
|
||||
if !validFuncName.MatchString(req.Name) || len(req.Name) > timeTriggerMaxNameLen {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "invalid timetrigger name: must match ^[a-z0-9]([a-z0-9-]*[a-z0-9])?$ and be <= 63 chars"}
|
||||
}
|
||||
if req.FunctionName == "" {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "functionName is required"}
|
||||
}
|
||||
if !validFuncName.MatchString(req.FunctionName) || len(req.FunctionName) > timeTriggerMaxNameLen {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "invalid functionName: must match ^[a-z0-9]([a-z0-9-]*[a-z0-9])?$ and be <= 63 chars"}
|
||||
}
|
||||
if req.Cron == "" {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "cron is required"}
|
||||
}
|
||||
|
||||
req.Method = strings.ToUpper(req.Method)
|
||||
if req.Method == "" {
|
||||
req.Method = timeTriggerDefaultMethod
|
||||
}
|
||||
if _, ok := validTimeTriggerMethods[req.Method]; !ok {
|
||||
return req, &apiErr{status: http.StatusBadRequest, message: "invalid method: must be GET, POST, PUT, DELETE or HEAD"}
|
||||
}
|
||||
|
||||
if req.SubPath == "" || req.SubPath == timeTriggerDefaultSubPath {
|
||||
req.SubPath = timeTriggerDefaultSubPath
|
||||
} else if !strings.HasPrefix(req.SubPath, "/") {
|
||||
req.SubPath = "/" + req.SubPath
|
||||
}
|
||||
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func (s *Server) ensureTimerWatchesNamespace(ctx context.Context, userNS string) error {
|
||||
userNS = strings.TrimSpace(userNS)
|
||||
if userNS == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
deployGVR := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
|
||||
deploy, err := s.dyn.Resource(deployGVR).Namespace(fissionSystemNamespace()).Get(ctx, "timer", metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return fmt.Errorf("get timer deployment: %w", err)
|
||||
}
|
||||
|
||||
containers, found, err := unstructured.NestedSlice(deploy.Object, "spec", "template", "spec", "containers")
|
||||
if err != nil {
|
||||
return fmt.Errorf("read timer containers: %w", err)
|
||||
}
|
||||
if !found {
|
||||
return fmt.Errorf("timer containers not found")
|
||||
}
|
||||
if len(containers) == 0 {
|
||||
return fmt.Errorf("timer containers empty")
|
||||
}
|
||||
container, ok := containers[0].(map[string]any)
|
||||
if !ok {
|
||||
return fmt.Errorf("timer container has unexpected shape")
|
||||
}
|
||||
envList, found, err := unstructured.NestedSlice(container, "env")
|
||||
if err != nil {
|
||||
return fmt.Errorf("read timer env: %w", err)
|
||||
}
|
||||
if !found {
|
||||
return fmt.Errorf("timer env not found")
|
||||
}
|
||||
|
||||
defaultNS := "default"
|
||||
resourceNamespaces := []string{"default"}
|
||||
defaultIdx := -1
|
||||
resourceIdx := -1
|
||||
for i, item := range envList {
|
||||
env, ok := item.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
name, _ := env["name"].(string)
|
||||
value, _ := env["value"].(string)
|
||||
switch name {
|
||||
case "FISSION_DEFAULT_NAMESPACE":
|
||||
defaultIdx = i
|
||||
if strings.TrimSpace(value) != "" {
|
||||
defaultNS = strings.TrimSpace(value)
|
||||
}
|
||||
case "FISSION_RESOURCE_NAMESPACES":
|
||||
resourceIdx = i
|
||||
resourceNamespaces = splitCSVNamespaces(value)
|
||||
}
|
||||
}
|
||||
|
||||
if len(resourceNamespaces) == 0 {
|
||||
resourceNamespaces = []string{defaultNS}
|
||||
}
|
||||
resourceNamespaces = appendNamespace(resourceNamespaces, userNS)
|
||||
resourceNamespaces = ensureDefaultFirst(resourceNamespaces, defaultNS)
|
||||
joined := strings.Join(resourceNamespaces, ",")
|
||||
|
||||
changed := false
|
||||
if defaultIdx >= 0 {
|
||||
env := envList[defaultIdx].(map[string]any)
|
||||
if env["value"] != defaultNS {
|
||||
env["value"] = defaultNS
|
||||
envList[defaultIdx] = env
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if resourceIdx >= 0 {
|
||||
env := envList[resourceIdx].(map[string]any)
|
||||
if env["value"] != joined {
|
||||
env["value"] = joined
|
||||
envList[resourceIdx] = env
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
|
||||
if !changed {
|
||||
return nil
|
||||
}
|
||||
container["env"] = envList
|
||||
containers[0] = container
|
||||
if err := unstructured.SetNestedSlice(deploy.Object, containers, "spec", "template", "spec", "containers"); err != nil {
|
||||
return fmt.Errorf("write timer env: %w", err)
|
||||
}
|
||||
if _, err := s.dyn.Resource(deployGVR).Namespace(fissionSystemNamespace()).Update(ctx, deploy, metav1.UpdateOptions{}); err != nil {
|
||||
return fmt.Errorf("update timer deployment: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func splitCSVNamespaces(raw string) []string {
|
||||
parts := strings.Split(raw, ",")
|
||||
seen := make(map[string]struct{}, len(parts))
|
||||
result := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
name := strings.TrimSpace(part)
|
||||
if name == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[name]; ok {
|
||||
continue
|
||||
}
|
||||
seen[name] = struct{}{}
|
||||
result = append(result, name)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func appendNamespace(namespaces []string, ns string) []string {
|
||||
for _, existing := range namespaces {
|
||||
if existing == ns {
|
||||
return namespaces
|
||||
}
|
||||
}
|
||||
return append(namespaces, ns)
|
||||
}
|
||||
|
||||
func ensureDefaultFirst(namespaces []string, defaultNS string) []string {
|
||||
uniq := splitCSVNamespaces(strings.Join(namespaces, ","))
|
||||
others := make([]string, 0, len(uniq))
|
||||
for _, ns := range uniq {
|
||||
if ns != defaultNS {
|
||||
others = append(others, ns)
|
||||
}
|
||||
}
|
||||
sort.Strings(others)
|
||||
return append([]string{defaultNS}, others...)
|
||||
}
|
||||
|
||||
func timeTriggerResponse(trigger *unstructured.Unstructured) map[string]any {
|
||||
result := map[string]any{}
|
||||
if trigger == nil {
|
||||
return result
|
||||
}
|
||||
result["raw"] = trigger.Object
|
||||
result["name"] = trigger.GetName()
|
||||
result["namespace"] = trigger.GetNamespace()
|
||||
if cron, found, _ := unstructured.NestedString(trigger.Object, "spec", "cron"); found {
|
||||
result["cron"] = cron
|
||||
}
|
||||
if method, found, _ := unstructured.NestedString(trigger.Object, "spec", "method"); found {
|
||||
result["method"] = method
|
||||
}
|
||||
if subpath, found, _ := unstructured.NestedString(trigger.Object, "spec", "subpath"); found {
|
||||
result["subpath"] = subpath
|
||||
}
|
||||
if functionName, found, _ := unstructured.NestedString(trigger.Object, "spec", "functionref", "name"); found {
|
||||
result["function"] = functionName
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
type apiErr struct {
|
||||
status int
|
||||
message string
|
||||
}
|
||||
|
||||
func (e *apiErr) Error() string {
|
||||
if e == nil {
|
||||
return ""
|
||||
}
|
||||
return e.message
|
||||
}
|
||||
|
||||
func errStatus(err error) int {
|
||||
if err == nil {
|
||||
return http.StatusInternalServerError
|
||||
}
|
||||
if ae, ok := err.(*apiErr); ok && ae.status != 0 {
|
||||
return ae.status
|
||||
}
|
||||
return http.StatusInternalServerError
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"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"
|
||||
runtime "k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
dynamicfake "k8s.io/client-go/dynamic/fake"
|
||||
)
|
||||
|
||||
func TestHandleTimeTriggersRootCreateListGetUpdateDelete(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
scheme := runtime.NewScheme()
|
||||
client := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(
|
||||
scheme,
|
||||
map[schema.GroupVersionResource]string{
|
||||
fission.TimeTrigGVR: "TimeTriggerList",
|
||||
},
|
||||
functionObject("fission-test", "hello"),
|
||||
)
|
||||
s := &Server{dyn: client, ns: "fission-test"}
|
||||
|
||||
createReq := httptest.NewRequest(http.MethodPost, "/console/api/timetriggers", strings.NewReader(`{"name":"cron-job","functionName":"hello","cron":"*/5 * * * *","method":"get","subpath":"logs"}`))
|
||||
createRec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersRoot(createRec, createReq)
|
||||
if createRec.Code != http.StatusCreated {
|
||||
t.Fatalf("create status = %d, want %d body=%s", createRec.Code, http.StatusCreated, createRec.Body.String())
|
||||
}
|
||||
created := decodeMap(t, createRec.Body.Bytes())
|
||||
if created["name"] != "cron-job" {
|
||||
t.Fatalf("create name = %v", created["name"])
|
||||
}
|
||||
if created["cron"] != "*/5 * * * *" {
|
||||
t.Fatalf("create cron = %v", created["cron"])
|
||||
}
|
||||
if created["method"] != http.MethodGet {
|
||||
t.Fatalf("create method = %v", created["method"])
|
||||
}
|
||||
if created["subpath"] != "/logs" {
|
||||
t.Fatalf("create subpath = %v", created["subpath"])
|
||||
}
|
||||
|
||||
createdObj, err := client.Resource(fission.TimeTrigGVR).Namespace("fission-test").Get(ctx, "cron-job", metav1.GetOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("get created timetrigger: %v", err)
|
||||
}
|
||||
if cron, _, _ := unstructured.NestedString(createdObj.Object, "spec", "cron"); cron != "*/5 * * * *" {
|
||||
t.Fatalf("stored cron = %q", cron)
|
||||
}
|
||||
|
||||
listReq := httptest.NewRequest(http.MethodGet, "/console/api/timetriggers", nil)
|
||||
listRec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersRoot(listRec, listReq)
|
||||
if listRec.Code != http.StatusOK {
|
||||
t.Fatalf("list status = %d body=%s", listRec.Code, listRec.Body.String())
|
||||
}
|
||||
if !strings.Contains(listRec.Body.String(), "cron-job") {
|
||||
t.Fatalf("list body does not contain created trigger: %s", listRec.Body.String())
|
||||
}
|
||||
|
||||
getReq := httptest.NewRequest(http.MethodGet, "/console/api/timetriggers/cron-job", nil)
|
||||
getRec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersAction(getRec, getReq)
|
||||
if getRec.Code != http.StatusOK {
|
||||
t.Fatalf("get status = %d body=%s", getRec.Code, getRec.Body.String())
|
||||
}
|
||||
got := decodeMap(t, getRec.Body.Bytes())
|
||||
if got["function"] != "hello" {
|
||||
t.Fatalf("get function = %v", got["function"])
|
||||
}
|
||||
|
||||
updateReq := httptest.NewRequest(http.MethodPut, "/console/api/timetriggers/cron-job", strings.NewReader(`{"name":"cron-job","functionName":"hello","cron":"@hourly","method":"post","subpath":"/"}`))
|
||||
updateRec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersAction(updateRec, updateReq)
|
||||
if updateRec.Code != http.StatusOK {
|
||||
t.Fatalf("update status = %d body=%s", updateRec.Code, updateRec.Body.String())
|
||||
}
|
||||
updated, err := client.Resource(fission.TimeTrigGVR).Namespace("fission-test").Get(ctx, "cron-job", metav1.GetOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("get updated timetrigger: %v", err)
|
||||
}
|
||||
if cron, _, _ := unstructured.NestedString(updated.Object, "spec", "cron"); cron != "@hourly" {
|
||||
t.Fatalf("updated cron = %q", cron)
|
||||
}
|
||||
|
||||
deleteReq := httptest.NewRequest(http.MethodDelete, "/console/api/timetriggers/cron-job", nil)
|
||||
deleteRec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersAction(deleteRec, deleteReq)
|
||||
if deleteRec.Code != http.StatusOK {
|
||||
t.Fatalf("delete status = %d body=%s", deleteRec.Code, deleteRec.Body.String())
|
||||
}
|
||||
if _, err := client.Resource(fission.TimeTrigGVR).Namespace("fission-test").Get(ctx, "cron-job", metav1.GetOptions{}); !apierrors.IsNotFound(err) {
|
||||
t.Fatalf("expected timetrigger to be deleted, got err=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleCreateTimeTriggerRejectsMissingFunction(t *testing.T) {
|
||||
client := dynamicfake.NewSimpleDynamicClient(runtime.NewScheme())
|
||||
s := &Server{dyn: client, ns: "fission-test"}
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/console/api/timetriggers", strings.NewReader(`{"name":"cron-job","functionName":"missing","cron":"*/5 * * * *"}`))
|
||||
rec := httptest.NewRecorder()
|
||||
s.handleTimeTriggersRoot(rec, req)
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want %d body=%s", rec.Code, http.StatusBadRequest, rec.Body.String())
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "not found") {
|
||||
t.Fatalf("expected not found error, got %s", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func decodeMap(t *testing.T, data []byte) map[string]any {
|
||||
t.Helper()
|
||||
var result map[string]any
|
||||
if err := json.Unmarshal(data, &result); err != nil {
|
||||
t.Fatalf("decode json: %v body=%s", err, string(data))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func functionObject(namespace, name string) *unstructured.Unstructured {
|
||||
return &unstructured.Unstructured{Object: map[string]any{
|
||||
"apiVersion": "fission.io/v1",
|
||||
"kind": "Function",
|
||||
"metadata": map[string]any{
|
||||
"name": name,
|
||||
"namespace": namespace,
|
||||
},
|
||||
}}
|
||||
}
|
||||
Reference in New Issue
Block a user