Merge pull request #163 from fnproject/hot-env-changes

change hot containers when route/app vars change

api/runner/task.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/docker/cli/cli/config/configfile"
 	"github.com/fnproject/fn/api/runner/drivers"
+	"github.com/fnproject/fn/api/runner/protocol"
 	"github.com/fnproject/fn/api/runner/task"
 	docker "github.com/fsouza/go-dockerclient"
 )
@@ -62,6 +63,9 @@ type containerTask struct {
 func (t *containerTask) Command() string { return "" }
 
 func (t *containerTask) EnvVars() map[string]string {
+	if protocol.IsStreamable(protocol.Protocol(t.cfg.Format)) {
+		return t.cfg.BaseEnv
+	}
 	return t.cfg.Env
 }
 func (t *containerTask) Input() io.Reader {
@@ -82,9 +86,8 @@ func (t *containerTask) IdleTimeout() time.Duration     { return t.cfg.IdleTimeo
 func (t *containerTask) Logger() (io.Writer, io.Writer) { return t.cfg.Stdout, t.cfg.Stderr }
 func (t *containerTask) Volumes() [][2]string           { return [][2]string{} }
 func (t *containerTask) WorkDir() string                { return "" }
-
+func (t *containerTask) Close()                         {}
-func (t *containerTask) Close()                 {}
+func (t *containerTask) WriteStat(drivers.Stat)         {}
-func (t *containerTask) WriteStat(drivers.Stat) {}
 
 // Implementing the docker.AuthConfiguration interface.  Pulling in
 // the docker repo password from environment variables

api/runner/task/task.go

@@ -16,7 +16,8 @@ type Config struct {
 	IdleTimeout  time.Duration
 	AppName      string
 	Memory       uint64
-	Env          map[string]string
+	BaseEnv      map[string]string // only app & route config vals [for hot]
+	Env          map[string]string // includes BaseEnv
 	Format       string
 	ReceivedTime time.Time
 	// Ready is used to await the first pull

api/runner/worker.go

@@ -2,6 +2,7 @@ package runner
 
 import (
 	"context"
+	"crypto/sha1"
 	"errors"
 	"fmt"
 	"io"
@@ -132,8 +133,8 @@ func (h *htfnmgr) getPipe(ctx context.Context, rnr *Runner, cfg *task.Config) ch
 		h.RLock()
 	}
 
-	// TODO(ccirello): re-implement this without memory allocation (fmt.Sprint)
+	fn := key(cfg)
-	fn := fmt.Sprint(cfg.AppName, ",", cfg.Path, cfg.Image, cfg.Timeout, cfg.Memory, cfg.Format)
+
 	svr, ok := h.hc[fn]
 	h.RUnlock()
 	if !ok {
@@ -149,6 +150,28 @@ func (h *htfnmgr) getPipe(ctx context.Context, rnr *Runner, cfg *task.Config) ch
 	return svr.tasksin
 }
 
+func key(cfg *task.Config) string {
+	// TODO we should probably colocate this with Config, but it's kind of hot
+	// specific so it makes sense here, too (just brittle & hidden)
+
+	// return a sha1 hash of a (hopefully) unique string of all the config
+	// values, to make map lookups quicker [than the giant unique string]
+	hash := sha1.New()
+	fmt.Fprint(hash, cfg.AppName, "\x00")
+	fmt.Fprint(hash, cfg.Path, "\x00")
+	fmt.Fprint(hash, cfg.Image, "\x00")
+	for k, v := range cfg.BaseEnv {
+		fmt.Fprint(hash, k, "\x00", v, "\x00")
+	}
+	fmt.Fprint(hash, cfg.Timeout, "\x00")
+	fmt.Fprint(hash, cfg.IdleTimeout, "\x00")
+	fmt.Fprint(hash, cfg.Memory, "\x00")
+	fmt.Fprint(hash, cfg.Format, "\x00")
+
+	var buf [sha1.Size]byte
+	return string(hash.Sum(buf[:]))
+}
+
 // htfnsvr is part of htfnmgr, abstracted apart for simplicity, its only
 // purpose is to test for hot functions saturation and try starting as many as
 // needed. In case of absence of workload, it will stop trying to start new hot
@@ -350,7 +373,6 @@ func (hc *htfn) serve(ctx context.Context) {
 		}
 	}()
 
-	cfg.Env["FN_FORMAT"] = cfg.Format
 	cfg.Timeout = 0 // add a timeout to simulate ab.end. failure.
 	cfg.Stdin = hc.containerIn
 	cfg.Stdout = hc.containerOut

api/server/runner.go

@@ -142,28 +142,39 @@ func (s *Server) serve(ctx context.Context, c *gin.Context, appName string, rout
 	if route.Format == "" {
 		route.Format = "default"
 	}
-	envVars := map[string]string{
+
-		"METHOD":   c.Request.Method,
+	// baseVars are the vars on the route & app, not on this specific request [for hot functions]
-		"APP_NAME": appName,
+	baseVars := make(map[string]string, len(app.Config)+len(route.Config)+3)
-		"ROUTE":    route.Path,
+	baseVars["FN_FORMAT"] = route.Format
-		"REQUEST_URL": fmt.Sprintf("%v//%v%v", func() string {
+	baseVars["APP_NAME"] = appName
-			if c.Request.TLS == nil {
+	baseVars["ROUTE"] = route.Path
-				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
+		k = toEnvName("", k)
+		baseVars[k] = v
 	}
 	for k, v := range route.Config {
-		envVars[toEnvName("", k)] = v
+		k = toEnvName("", k)
+		baseVars[k] = v
 	}
 
+	// envVars contains the full set of env vars, per request + base
+	envVars := make(map[string]string, len(baseVars)+len(params)+len(c.Request.Header)+3)
+
+	for k, v := range baseVars {
+		envVars[k] = v
+	}
+
+	envVars["CALL_ID"] = reqID
+	envVars["METHOD"] = c.Request.Method
+	envVars["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())
+
 	// params
 	for _, param := range params {
 		envVars[toEnvName("PARAM", param.Key)] = param.Value
@@ -177,6 +188,7 @@ func (s *Server) serve(ctx context.Context, c *gin.Context, appName string, rout
 	cfg := &task.Config{
 		AppName:      appName,
 		Path:         route.Path,
+		BaseEnv:      baseVars,
 		Env:          envVars,
 		Format:       route.Format,
 		ID:           reqID,