mirror of
https://github.com/fnproject/fn.git
synced 2022-10-28 21:29:17 +03:00
308 lines
7.6 KiB
Go
308 lines
7.6 KiB
Go
package server
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"io/ioutil"
|
|
"net/http"
|
|
"path"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/Sirupsen/logrus"
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/go-openapi/strfmt"
|
|
"gitlab-odx.oracle.com/odx/functions/api"
|
|
"gitlab-odx.oracle.com/odx/functions/api/id"
|
|
"gitlab-odx.oracle.com/odx/functions/api/models"
|
|
"gitlab-odx.oracle.com/odx/functions/api/runner"
|
|
"gitlab-odx.oracle.com/odx/functions/api/runner/common"
|
|
"gitlab-odx.oracle.com/odx/functions/api/runner/task"
|
|
)
|
|
|
|
type runnerResponse struct {
|
|
RequestID string `json:"request_id,omitempty"`
|
|
Error *models.ErrorBody `json:"error,omitempty"`
|
|
}
|
|
|
|
func (s *Server) handleSpecial(c *gin.Context) {
|
|
ctx := c.Request.Context()
|
|
|
|
ctx = context.WithValue(ctx, api.AppName, "")
|
|
c.Set(api.AppName, "")
|
|
ctx = context.WithValue(ctx, api.Path, c.Request.URL.Path)
|
|
c.Set(api.Path, c.Request.URL.Path)
|
|
|
|
r, err := s.UseSpecialHandlers(c.Writer, c.Request)
|
|
if err != nil {
|
|
handleErrorResponse(c, err)
|
|
return
|
|
}
|
|
|
|
c.Request = r
|
|
c.Set(api.AppName, r.Context().Value(api.AppName).(string))
|
|
if c.MustGet(api.AppName).(string) == "" {
|
|
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
|
return
|
|
}
|
|
|
|
// now call the normal runner call
|
|
s.handleRequest(c, nil)
|
|
}
|
|
|
|
func toEnvName(envtype, name string) string {
|
|
name = strings.ToUpper(strings.Replace(name, "-", "_", -1))
|
|
if envtype == "" {
|
|
return name
|
|
}
|
|
return fmt.Sprintf("%s_%s", envtype, name)
|
|
}
|
|
|
|
func (s *Server) handleRequest(c *gin.Context, enqueue models.Enqueue) {
|
|
if strings.HasPrefix(c.Request.URL.Path, "/v1") {
|
|
c.Status(http.StatusNotFound)
|
|
return
|
|
}
|
|
|
|
ctx := c.Request.Context()
|
|
|
|
reqID := id.New().String()
|
|
ctx, log := common.LoggerWithFields(ctx, logrus.Fields{"call_id": reqID})
|
|
|
|
var err error
|
|
var payload io.Reader
|
|
|
|
if c.Request.Method == "POST" {
|
|
payload = c.Request.Body
|
|
// Load complete body and close
|
|
defer func() {
|
|
io.Copy(ioutil.Discard, c.Request.Body)
|
|
c.Request.Body.Close()
|
|
}()
|
|
} else if c.Request.Method == "GET" {
|
|
reqPayload := c.Request.URL.Query().Get("payload")
|
|
payload = strings.NewReader(reqPayload)
|
|
}
|
|
|
|
r, routeExists := c.Get(api.Path)
|
|
if !routeExists {
|
|
r = "/"
|
|
}
|
|
|
|
reqRoute := &models.Route{
|
|
AppName: c.MustGet(api.AppName).(string),
|
|
Path: path.Clean(r.(string)),
|
|
}
|
|
|
|
s.FireBeforeDispatch(ctx, reqRoute)
|
|
|
|
appName := reqRoute.AppName
|
|
path := reqRoute.Path
|
|
|
|
app, err := s.Datastore.GetApp(ctx, appName)
|
|
if err != nil {
|
|
handleErrorResponse(c, err)
|
|
return
|
|
} else if app == nil {
|
|
handleErrorResponse(c, models.ErrAppsNotFound)
|
|
return
|
|
}
|
|
|
|
log.WithFields(logrus.Fields{"app": appName, "path": path}).Debug("Finding route on datastore")
|
|
routes, err := s.loadroutes(ctx, models.RouteFilter{AppName: appName, Path: path})
|
|
if err != nil {
|
|
handleErrorResponse(c, err)
|
|
return
|
|
}
|
|
|
|
if len(routes) == 0 {
|
|
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
|
return
|
|
}
|
|
|
|
log.WithField("routes", len(routes)).Debug("Got routes from datastore")
|
|
route := routes[0]
|
|
log = log.WithFields(logrus.Fields{"app": appName, "path": route.Path, "image": route.Image})
|
|
|
|
if s.serve(ctx, c, appName, route, app, path, reqID, payload, enqueue) {
|
|
s.FireAfterDispatch(ctx, reqRoute)
|
|
return
|
|
}
|
|
|
|
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
|
}
|
|
|
|
func (s *Server) loadroutes(ctx context.Context, filter models.RouteFilter) ([]*models.Route, error) {
|
|
if route, ok := s.cacheget(filter.AppName, filter.Path); ok {
|
|
return []*models.Route{route}, nil
|
|
}
|
|
resp, err := s.singleflight.do(
|
|
filter,
|
|
func() (interface{}, error) {
|
|
return s.Datastore.GetRoutesByApp(ctx, filter.AppName, &filter)
|
|
},
|
|
)
|
|
return resp.([]*models.Route), err
|
|
}
|
|
|
|
// TODO: Should remove *gin.Context from these functions, should use only context.Context
|
|
func (s *Server) serve(ctx context.Context, c *gin.Context, appName string, route *models.Route, app *models.App, path, reqID string, payload io.Reader, enqueue models.Enqueue) (ok bool) {
|
|
ctx, log := common.LoggerWithFields(ctx, logrus.Fields{"app": appName, "route": route.Path, "image": route.Image})
|
|
|
|
params, match := matchRoute(route.Path, path)
|
|
if !match {
|
|
return false
|
|
}
|
|
|
|
var stdout bytes.Buffer // TODO: should limit the size of this, error if gets too big. akin to: https://golang.org/pkg/io/#LimitReader
|
|
|
|
if route.Format == "" {
|
|
route.Format = "default"
|
|
}
|
|
envVars := map[string]string{
|
|
"METHOD": c.Request.Method,
|
|
"APP_NAME": appName,
|
|
"ROUTE": route.Path,
|
|
"REQUEST_URL": fmt.Sprintf("%v//%v%v", func() string {
|
|
if c.Request.TLS == nil {
|
|
return "http"
|
|
}
|
|
return "https"
|
|
}(), c.Request.Host, c.Request.URL.String()),
|
|
"CALL_ID": reqID,
|
|
"FORMAT": route.Format,
|
|
}
|
|
|
|
// app config
|
|
for k, v := range app.Config {
|
|
envVars[toEnvName("", k)] = v
|
|
}
|
|
for k, v := range route.Config {
|
|
envVars[toEnvName("", k)] = v
|
|
}
|
|
|
|
// params
|
|
for _, param := range params {
|
|
envVars[toEnvName("PARAM", param.Key)] = param.Value
|
|
}
|
|
|
|
// headers
|
|
for header, value := range c.Request.Header {
|
|
envVars[toEnvName("HEADER", header)] = strings.Join(value, " ")
|
|
}
|
|
|
|
cfg := &task.Config{
|
|
AppName: appName,
|
|
Path: route.Path,
|
|
Env: envVars,
|
|
Format: route.Format,
|
|
ID: reqID,
|
|
Image: route.Image,
|
|
Memory: route.Memory,
|
|
Stdin: payload,
|
|
Stdout: &stdout,
|
|
Timeout: time.Duration(route.Timeout) * time.Second,
|
|
IdleTimeout: time.Duration(route.IdleTimeout) * time.Second,
|
|
ReceivedTime: time.Now(),
|
|
Ready: make(chan struct{}),
|
|
}
|
|
|
|
// ensure valid values
|
|
if cfg.Timeout <= 0 {
|
|
cfg.Timeout = runner.DefaultTimeout
|
|
}
|
|
if cfg.IdleTimeout <= 0 {
|
|
cfg.IdleTimeout = runner.DefaultIdleTimeout
|
|
}
|
|
|
|
s.Runner.Enqueue()
|
|
createdAt := strfmt.DateTime(time.Now())
|
|
newTask := &models.Task{}
|
|
newTask.Image = &cfg.Image
|
|
newTask.ID = cfg.ID
|
|
newTask.CreatedAt = createdAt
|
|
newTask.Path = route.Path
|
|
newTask.EnvVars = cfg.Env
|
|
newTask.AppName = cfg.AppName
|
|
|
|
switch route.Type {
|
|
case "async":
|
|
// Read payload
|
|
pl, err := ioutil.ReadAll(cfg.Stdin)
|
|
if err != nil {
|
|
handleErrorResponse(c, models.ErrInvalidPayload)
|
|
return true
|
|
}
|
|
// Create Task
|
|
priority := int32(0)
|
|
newTask.Priority = &priority
|
|
newTask.Payload = string(pl)
|
|
// Push to queue
|
|
enqueue(c, s.MQ, newTask)
|
|
log.Info("Added new task to queue")
|
|
c.JSON(http.StatusAccepted, map[string]string{"call_id": newTask.ID})
|
|
|
|
default:
|
|
result, err := s.Runner.RunTrackedTask(newTask, ctx, cfg)
|
|
if result != nil {
|
|
waitTime := result.StartTime().Sub(cfg.ReceivedTime)
|
|
c.Header("XXX-FXLB-WAIT", waitTime.String())
|
|
}
|
|
|
|
if err != nil {
|
|
c.JSON(http.StatusInternalServerError, runnerResponse{
|
|
RequestID: cfg.ID,
|
|
Error: &models.ErrorBody{
|
|
Message: err.Error(),
|
|
},
|
|
})
|
|
log.WithError(err).Error("Failed to run task")
|
|
break
|
|
}
|
|
|
|
for k, v := range route.Headers {
|
|
c.Header(k, v[0])
|
|
}
|
|
|
|
// this will help users to track sync execution in a manner of async
|
|
// FN_CALL_ID is an equivalent of call_id
|
|
c.Header("FN_CALL_ID", newTask.ID)
|
|
|
|
switch result.Status() {
|
|
case "success":
|
|
c.Data(http.StatusOK, "", stdout.Bytes())
|
|
case "timeout":
|
|
c.JSON(http.StatusGatewayTimeout, runnerResponse{
|
|
RequestID: cfg.ID,
|
|
Error: &models.ErrorBody{
|
|
Message: models.ErrRunnerTimeout.Error(),
|
|
},
|
|
})
|
|
default:
|
|
c.JSON(http.StatusInternalServerError, runnerResponse{
|
|
RequestID: cfg.ID,
|
|
Error: &models.ErrorBody{
|
|
Message: result.Error(),
|
|
},
|
|
})
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
var fakeHandler = func(http.ResponseWriter, *http.Request, Params) {}
|
|
|
|
func matchRoute(baseRoute, route string) (Params, bool) {
|
|
tree := &node{}
|
|
tree.addRoute(baseRoute, fakeHandler)
|
|
handler, p, _ := tree.getValue(route)
|
|
if handler == nil {
|
|
return nil, false
|
|
}
|
|
|
|
return p, true
|
|
}
|