package agent import ( "context" "errors" "fmt" "io" "net/http" "strconv" "strings" "time" "github.com/fnproject/fn/api/agent/drivers" "github.com/fnproject/fn/api/common" "github.com/fnproject/fn/api/id" "github.com/fnproject/fn/api/models" "github.com/go-openapi/strfmt" "github.com/opentracing/opentracing-go" "github.com/sirupsen/logrus" ) type Call interface { // Model will return the underlying models.Call configuration for this call. // TODO we could respond to async correctly from agent but layering, this // is only because the front end has different responses based on call type. // try to discourage use elsewhere until this gets pushed down more... Model() *models.Call // Start will be called before this call is executed, it may be used to // guarantee mutual exclusion, check docker permissions, update timestamps, // etc. // TODO Start and End can likely be unexported as they are only used in the agent, // and on a type which is constructed in a specific agent. meh. Start(ctx context.Context) error // End will be called immediately after attempting a call execution, // regardless of whether the execution failed or not. An error will be passed // to End, which if nil indicates a successful execution. Any error returned // from End will be returned as the error from Submit. End(ctx context.Context, err error) error } // TODO build w/o closures... lazy type CallOpt func(a *agent, c *call) error type Param struct { Key string Value string } type Params []Param func fixupRequestURL(req *http.Request) string { if req.URL.Scheme == "" { if req.TLS == nil { req.URL.Scheme = "http" } else { req.URL.Scheme = "https" } } if req.URL.Host == "" { req.URL.Host = req.Host } return req.URL.String() } func FromRequest(appName, path string, req *http.Request, params Params) CallOpt { return func(a *agent, c *call) error { app, err := a.da.GetApp(req.Context(), appName) if err != nil { return err } route, err := a.da.GetRoute(req.Context(), appName, path) if err != nil { return err } if route.Format == "" { route.Format = models.FormatDefault } url := fixupRequestURL(req) id := id.New().String() // baseVars are the vars on the route & app, not on this specific request [for hot functions] baseVars := make(map[string]string, len(app.Config)+len(route.Config)+3) // add app & route config before our standard additions for k, v := range app.Config { k = toEnvName("", k) baseVars[k] = v } for k, v := range route.Config { k = toEnvName("", k) baseVars[k] = v } baseVars["FN_FORMAT"] = route.Format baseVars["FN_APP_NAME"] = appName baseVars["FN_PATH"] = route.Path // TODO: might be a good idea to pass in: envVars["FN_BASE_PATH"] = fmt.Sprintf("/r/%s", appName) || "/" if using DNS entries per app baseVars["FN_MEMORY"] = fmt.Sprintf("%d", route.Memory) baseVars["FN_TYPE"] = route.Type // envVars contains the full set of env vars, per request + base envVars := make(map[string]string, len(baseVars)+len(params)+len(req.Header)+3) for k, v := range baseVars { envVars[k] = v } envVars["FN_CALL_ID"] = id envVars["FN_METHOD"] = req.Method envVars["FN_REQUEST_URL"] = url // params for _, param := range params { envVars[toEnvName("FN_PARAM", param.Key)] = param.Value } headerVars := make(map[string]string, len(req.Header)) for k, v := range req.Header { if !noOverrideVars(k) { // NOTE if we don't do this, they'll leak in (don't want people relying on this behavior) headerVars[toEnvName("FN_HEADER", k)] = strings.Join(v, ", ") } } // add all the env vars we build to the request headers for k, v := range envVars { if noOverrideVars(k) { // overwrite the passed in request headers explicitly with the generated ones req.Header.Set(k, v) } else { req.Header.Add(k, v) } } for k, v := range headerVars { envVars[k] = v } // TODO this relies on ordering of opts, but tests make sure it works, probably re-plumb/destroy headers // TODO async should probably supply an http.ResponseWriter that records the logs, to attach response headers to if rw, ok := c.w.(http.ResponseWriter); ok { rw.Header().Add("FN_CALL_ID", id) for k, vs := range route.Headers { for _, v := range vs { // pre-write in these headers to response rw.Header().Add(k, v) } } } // this ensures that there is an image, path, timeouts, memory, etc are valid. // NOTE: this means assign any changes above into route's fields err = route.Validate() if err != nil { return err } c.Call = &models.Call{ ID: id, AppName: appName, Path: route.Path, Image: route.Image, // Delay: 0, Type: route.Type, Format: route.Format, // Payload: TODO, Priority: new(int32), // TODO this is crucial, apparently Timeout: route.Timeout, IdleTimeout: route.IdleTimeout, Memory: route.Memory, BaseEnv: baseVars, EnvVars: envVars, CreatedAt: strfmt.DateTime(time.Now()), URL: url, Method: req.Method, } c.req = req return nil } } func noOverrideVars(key string) bool { // descrepency in casing b/w req headers and env vars, force matches return overrideVars[strings.ToUpper(key)] } // overrideVars means that the app config, route config or header vars // must not overwrite the generated values in call construction. var overrideVars = map[string]bool{ "FN_FORMAT": true, "FN_APP_NAME": true, "FN_PATH": true, "FN_MEMORY": true, "FN_TYPE": true, "FN_CALL_ID": true, "FN_METHOD": true, "FN_REQUEST_URL": true, } // TODO this currently relies on FromRequest having happened before to create the model // here, to be a fully qualified model. We probably should double check but having a way // to bypass will likely be what's used anyway unless forced. func FromModel(mCall *models.Call) CallOpt { return func(a *agent, c *call) error { c.Call = mCall req, err := http.NewRequest(c.Method, c.URL, strings.NewReader(c.Payload)) if err != nil { return err } // HACK: only applies to async below, for async we need to restore // content-length and content-type of the original request, which are // derived from Payload and original content-type which now is in // Fn_header_content_type if c.Type == models.TypeAsync { // Hoist original request content type into async request if req.Header.Get("Content-Type") == "" { content_type, ok := c.EnvVars["Fn_header_content_type"] if ok { req.Header.Set("Content-Type", content_type) } } // Ensure content-length in async requests for protocol/http DumpRequestTo() if req.ContentLength == -1 || req.Header.Get("Content-Length") == "" { req.ContentLength = int64(len(c.Payload)) req.Header.Set("Content-Length", strconv.FormatInt(int64(len(c.Payload)), 10)) } } for k, v := range c.EnvVars { // TODO if we don't store env as []string headers are messed up req.Header.Set(k, v) } c.req = req // TODO anything else really? return nil } } // TODO this should be required func WithWriter(w io.Writer) CallOpt { return func(a *agent, c *call) error { c.w = w return nil } } // GetCall builds a Call that can be used to submit jobs to the agent. // // TODO we could make this package level just moving the cache around. meh. // TODO where to put this? async and sync both call this func (a *agent) GetCall(opts ...CallOpt) (Call, error) { var c call for _, o := range opts { err := o(a, &c) if err != nil { return nil, err } } // TODO typed errors to test if c.req == nil || c.Call == nil { return nil, errors.New("no model or request provided for call") } c.da = a.da c.ct = a ctx, _ := common.LoggerWithFields(c.req.Context(), logrus.Fields{"id": c.ID, "app": c.AppName, "route": c.Path}) c.req = c.req.WithContext(ctx) // setup stderr logger separate (don't inherit ctx vars) logger := logrus.WithFields(logrus.Fields{"user_log": true, "app_name": c.AppName, "path": c.Path, "image": c.Image, "call_id": c.ID}) c.stderr = setupLogger(logger) if c.w == nil { // send STDOUT to logs if no writer given (async...) // TODO we could/should probably make this explicit to GetCall, ala 'WithLogger', but it's dupe code (who cares?) c.w = c.stderr } now := time.Now() slotDeadline := now.Add(time.Duration(c.Call.Timeout) * time.Second / 2) execDeadline := now.Add(time.Duration(c.Call.Timeout) * time.Second) c.slotDeadline = slotDeadline c.execDeadline = execDeadline execDeadlineStr := strfmt.DateTime(execDeadline).String() c.EnvVars["FN_DEADLINE"] = execDeadlineStr c.req.Header.Set("FN_DEADLINE", execDeadlineStr) return &c, nil } type call struct { *models.Call da DataAccess w io.Writer req *http.Request stderr io.ReadWriteCloser ct callTrigger slots *slotQueue slotDeadline time.Time execDeadline time.Time } func (c *call) Model() *models.Call { return c.Call } func (c *call) Start(ctx context.Context) error { span, ctx := opentracing.StartSpanFromContext(ctx, "agent_call_start") defer span.Finish() // Check context timeouts, errors if ctx.Err() != nil { return ctx.Err() } c.StartedAt = strfmt.DateTime(time.Now()) c.Status = "running" if rw, ok := c.w.(http.ResponseWriter); ok { // TODO need to figure out better way to wire response headers in rw.Header().Set("XXX-FXLB-WAIT", time.Time(c.StartedAt).Sub(time.Time(c.CreatedAt)).String()) } if c.Type == models.TypeAsync { // XXX (reed): make sure MQ reservation is lengthy. to skirt MQ semantics, // we could add a new message to MQ w/ delay of call.Timeout and delete the // old one (in that order), after marking the call as running in the db // (see below) // XXX (reed): should we store the updated started_at + status? we could // use this so that if we pick up a call from mq and find its status is // running to avoid running the call twice and potentially mark it as // errored (built in long running task detector, so to speak...) err := c.da.Start(ctx, c.Model()) if err != nil { return err // let another thread try this } } err := c.ct.fireBeforeCall(ctx, c.Model()) if err != nil { return fmt.Errorf("BeforeCall: %v", err) } return nil } func (c *call) End(ctx context.Context, errIn error) error { span, ctx := opentracing.StartSpanFromContext(ctx, "agent_call_end") defer span.Finish() c.CompletedAt = strfmt.DateTime(time.Now()) switch errIn { case nil: c.Status = "success" case context.DeadlineExceeded: c.Status = "timeout" default: c.Status = "error" c.Error = errIn.Error() } // ensure stats histogram is reasonably bounded c.Call.Stats = drivers.Decimate(240, c.Call.Stats) if err := c.da.Finish(ctx, c.Model(), c.stderr, c.Type == models.TypeAsync); err != nil { common.Logger(ctx).WithError(err).Error("error finalizing call on datastore/mq") // note: Not returning err here since the job could have already finished successfully. } // NOTE call this after InsertLog or the buffer will get reset c.stderr.Close() if err := c.ct.fireAfterCall(ctx, c.Model()); err != nil { return fmt.Errorf("AfterCall: %v", err) } return errIn // original error, important for use in sync call returns } func toEnvName(envtype, name string) string { name = strings.Replace(name, "-", "_", -1) if envtype == "" { return name } return fmt.Sprintf("%s_%s", envtype, name) }