506 lines
16 KiB
Go
506 lines
16 KiB
Go
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
|
|
cronGatewayURL = "http://fission-console.fission.svc.cluster.local"
|
|
)
|
|
|
|
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 {
|
|
if apierrors.IsNotFound(err) {
|
|
return 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
|
|
routerIdx := -1
|
|
changed := false
|
|
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)
|
|
case "FISSION_ROUTER_URL":
|
|
routerIdx = i
|
|
if value != cronGatewayURL {
|
|
env["value"] = cronGatewayURL
|
|
envList[i] = env
|
|
changed = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if len(resourceNamespaces) == 0 {
|
|
resourceNamespaces = []string{defaultNS}
|
|
}
|
|
resourceNamespaces = appendNamespace(resourceNamespaces, userNS)
|
|
resourceNamespaces = ensureDefaultFirst(resourceNamespaces, defaultNS)
|
|
joined := strings.Join(resourceNamespaces, ",")
|
|
|
|
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 resourceIdx < 0 {
|
|
envList = append(envList, map[string]any{"name": "FISSION_RESOURCE_NAMESPACES", "value": joined})
|
|
changed = true
|
|
}
|
|
if defaultIdx < 0 {
|
|
envList = append(envList, map[string]any{"name": "FISSION_DEFAULT_NAMESPACE", "value": defaultNS})
|
|
changed = true
|
|
}
|
|
if routerIdx < 0 {
|
|
envList = append(envList, map[string]any{"name": "FISSION_ROUTER_URL", "value": cronGatewayURL})
|
|
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
|
|
}
|