HTTP Triggers hookup (#1086)

* Initial suypport for invoking tiggers * dupe method * tighten server constraints * runner tests not working yet * basic route tests passing * post rebase fixes * add hybrid support for trigger invoke and tests * consoloidate all hybrid evil into one place * cleanup and make triggers unique by source * fix oops with Agent * linting * review fixes
2022-10-28 21:29:17 +03:00 · 2018-07-05 18:56:07 +01:00
parent b07a000a18
commit b8b544ed25
38 changed files with 2208 additions and 865 deletions
--- a/api/agent/call.go
+++ b/api/agent/call.go
@@ -46,23 +46,15 @@ type CallOverrider func(*models.Call, map[string]string) (map[string]string, err
 // TODO build w/o closures... lazy
 type CallOpt func(c *call) error

-type Param struct {
-	Key   string
-	Value string
-}
-type Params []Param
-
 const (
 	ceMimeType = "application/cloudevents+json"
 )

-func FromRequest(a Agent, app *models.App, path string, req *http.Request) CallOpt {
+// FromRequest initialises a call to a route from an HTTP request
+// deprecate with routes
+func FromRequest(app *models.App, route *models.Route, req *http.Request) CallOpt {
 	return func(c *call) error {
 		ctx := req.Context()
-		route, err := a.GetRoute(ctx, app.ID, path)
-		if err != nil {
-			return err
-		}

 		log := common.Logger(ctx)
 		// Check whether this is a CloudEvent, if coming in via HTTP router (only way currently), then we'll look for a special header
@@ -125,7 +117,7 @@ func FromRequest(a Agent, app *models.App, path string, req *http.Request) CallO
 			Memory:      route.Memory,
 			CPUs:        route.CPUs,
 			Config:      buildConfig(app, route),
-			Annotations: buildAnnotations(app, route),
+			Annotations: app.Annotations.MergeChange(route.Annotations),
 			Headers:     req.Header,
 			CreatedAt:   common.DateTime(time.Now()),
 			URL:         reqURL(req),
@@ -139,6 +131,77 @@ func FromRequest(a Agent, app *models.App, path string, req *http.Request) CallO
 	}
 }

+// Sets up a call from an http trigger request
+func FromHTTPTriggerRequest(app *models.App, fn *models.Fn, trigger *models.Trigger, req *http.Request) CallOpt {
+	return func(c *call) error {
+		ctx := req.Context()
+
+		log := common.Logger(ctx)
+		// Check whether this is a CloudEvent, if coming in via HTTP router (only way currently), then we'll look for a special header
+		// Content-Type header: https://github.com/cloudevents/spec/blob/master/http-transport-binding.md#32-structured-content-mode
+		// Expected Content-Type for a CloudEvent: application/cloudevents+json; charset=UTF-8
+		contentType := req.Header.Get("Content-Type")
+		t, _, err := mime.ParseMediaType(contentType)
+		if err != nil {
+			// won't fail here, but log
+			log.Debugf("Could not parse Content-Type header: %v", err)
+		} else {
+			if t == ceMimeType {
+				c.IsCloudEvent = true
+				fn.Format = models.FormatCloudEvent
+			}
+		}
+
+		if fn.Format == "" {
+			fn.Format = models.FormatDefault
+		}
+
+		id := id.New().String()
+
+		// 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)
+		}
+
+		var syslogURL string
+		if app.SyslogURL != nil {
+			syslogURL = *app.SyslogURL
+		}
+
+		c.Call = &models.Call{
+			ID:    id,
+			Path:  trigger.Source,
+			Image: fn.Image,
+			// Delay: 0,
+			Type:   "sync",
+			Format: fn.Format,
+			// Payload: TODO,
+			Priority:    new(int32), // TODO this is crucial, apparently
+			Timeout:     fn.Timeout,
+			IdleTimeout: fn.IdleTimeout,
+			TmpFsSize:   0, // TODO clean up this
+			Memory:      fn.Memory,
+			CPUs:        0, // TODO clean up this
+			Config:      buildTriggerConfig(app, fn, trigger),
+			// TODO - this wasn't really the intention here (that annotations would naturally cascade
+			// but seems to be necessary for some runner behaviour
+			Annotations: app.Annotations.MergeChange(fn.Annotations).MergeChange(trigger.Annotations),
+			Headers:     req.Header,
+			CreatedAt:   common.DateTime(time.Now()),
+			URL:         reqURL(req),
+			Method:      req.Method,
+			AppID:       app.ID,
+			FnID:        fn.ID,
+			TriggerID:   trigger.ID,
+			SyslogURL:   syslogURL,
+		}
+
+		c.req = req
+		return nil
+	}
+}
+
 func buildConfig(app *models.App, route *models.Route) models.Config {
 	conf := make(models.Config, 8+len(app.Config)+len(route.Config))
 	for k, v := range app.Config {
@@ -163,15 +226,24 @@ func buildConfig(app *models.App, route *models.Route) models.Config {
 	return conf
 }

-func buildAnnotations(app *models.App, route *models.Route) models.Annotations {
-	ann := make(models.Annotations, len(app.Annotations)+len(route.Annotations))
-	for k, v := range app.Annotations {
-		ann[k] = v
+func buildTriggerConfig(app *models.App, fn *models.Fn, trigger *models.Trigger) models.Config {
+	conf := make(models.Config, 8+len(app.Config)+len(fn.Config))
+	for k, v := range app.Config {
+		conf[k] = v
 	}
-	for k, v := range route.Annotations {
-		ann[k] = v
+	for k, v := range fn.Config {
+		conf[k] = v
 	}
-	return ann
+
+	conf["FN_FORMAT"] = fn.Format
+	conf["FN_APP_NAME"] = app.Name
+	conf["FN_PATH"] = trigger.Source
+	// TODO: might be a good idea to pass in: "FN_BASE_PATH" = fmt.Sprintf("/r/%s", appName) || "/" if using DNS entries per app
+	conf["FN_MEMORY"] = fmt.Sprintf("%d", fn.Memory)
+	conf["FN_TYPE"] = "sync"
+	conf["FN_FN_ID"] = fn.ID
+
+	return conf
 }

 func reqURL(req *http.Request) string {
@@ -188,9 +260,7 @@ func reqURL(req *http.Request) string {
 	return req.URL.String()
 }

-// 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.
+// FromModel creates a call object from an existing stored call model object, reading the body from the stored call payload
 func FromModel(mCall *models.Call) CallOpt {
 	return func(c *call) error {
 		c.Call = mCall
@@ -207,6 +277,7 @@ func FromModel(mCall *models.Call) CallOpt {
 	}
 }

+// FromModelAndInput creates a call object from an existing stored call model object , reading the body from a provided stream
 func FromModelAndInput(mCall *models.Call, in io.ReadCloser) CallOpt {
 	return func(c *call) error {
 		c.Call = mCall
@@ -223,6 +294,7 @@ func FromModelAndInput(mCall *models.Call, in io.ReadCloser) CallOpt {
 	}
 }

+// WithWriter sets the writier that the call uses to send its output message to
 // TODO this should be required
 func WithWriter(w io.Writer) CallOpt {
 	return func(c *call) error {
@@ -231,6 +303,7 @@ func WithWriter(w io.Writer) CallOpt {
 	}
 }

+// WithContext overrides the context on the call
 func WithContext(ctx context.Context) CallOpt {
 	return func(c *call) error {
 		c.req = c.req.WithContext(ctx)
@@ -238,6 +311,7 @@ func WithContext(ctx context.Context) CallOpt {
 	}
 }

+// WithExtensions adds internal attributes to the call that can be interpreted by extensions in the agent
 // Pure runner can use this to pass an extension to the call
 func WithExtensions(extensions map[string]string) CallOpt {
 	return func(c *call) error {
@@ -287,7 +361,7 @@ func (a *agent) GetCall(opts ...CallOpt) (Call, error) {

 	setupCtx(&c)

-	c.da = a.da
+	c.handler = a.da
 	c.ct = a
 	c.stderr = setupLogger(c.req.Context(), a.cfg.MaxLogSize, c.Call)
 	if c.w == nil {
@@ -321,7 +395,7 @@ type call struct {
 	// IsCloudEvent flag whether this was ingested as a cloud event. This may become the default or only way.
 	IsCloudEvent bool `json:"is_cloud_event"`

-	da             DataAccess
+	handler        CallHandler
 	w              io.Writer
 	req            *http.Request
 	stderr         io.ReadWriteCloser
@@ -336,6 +410,8 @@ type call struct {
 	extensions map[string]string
 }

+// SlotHashId returns a string identity for this call that can be used to uniquely place the call in a given container
+// This should correspond to a unique identity (including data changes) of the underlying function
 func (c *call) SlotHashId() string {
 	return c.slotHashId
 }
@@ -393,7 +469,7 @@ func (c *call) Start(ctx context.Context) error {
 		// 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())
+		err := c.handler.Start(ctx, c.Model())
 		if err != nil {
 			return err // let another thread try this
 		}
@@ -426,7 +502,7 @@ func (c *call) End(ctx context.Context, errIn error) 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 {
+	if err := c.handler.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.
 	}