Add tmpfs IOFS (#1212)

* Define an interface for IOFS handling. Add no-op and temporary directory implementations.

* Move IOFS stuff out into separate file, add basic tmpfs implementation for linux only

* Switch between directory and tmpfs based on platform and config

* Respect FN_IOFS_OPTS

* Make directory iofs default on all platforms

* At least try to clean up a bit on failure

* Add backout if IOFS creation fails

* Add comment about iofs.Close

api/agent/agent.go

@@ -6,10 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
-	"os"
 	"strings"
 	"sync"
 	"time"
@@ -834,6 +832,7 @@ func (a *agent) prepCold(ctx context.Context, call *call, tok ResourceToken, ch
 		memory:  call.Memory,
 		cpus:    uint64(call.CPUs),
 		fsSize:  a.cfg.MaxFsSize,
+		iofs:    &noopIOFS{},
 		timeout: time.Duration(call.Timeout) * time.Second, // this is unnecessary, but in case removal fails...
 		logCfg: drivers.LoggerConfig{
 			URL: strings.TrimSpace(call.SyslogURL),
@@ -1007,49 +1006,6 @@ func (a *agent) runHot(ctx context.Context, call *call, tok ResourceToken, state
 	logger.WithError(res.Error()).Info("hot function terminated")
 }
 
-// Creates an IO directory for  container  sockets  - returns a pair of directories
-// first is the directory relative to the agent (what I watch and talk to), second is a directory relative to docker (what I ask docker to mount)
-// e.g. If IOFSAgentPath is set /data/iofs/ and IOFSMountRoot is set to /my/path/to/iofs/ this will return paths like:
-// /data/iofs/iofs829027/  /my/path/to/iofs/iofs829027/ respectively
-// If either IOFSAgentPath is unset it will always return paths relative to /tmp/ on the docker host and
-// if only IOFSMountPath is unset it will return the same directory for both
-func createIOFS(cfg *Config) (string, string, error) {
-	// XXX(reed): need to ensure these are cleaned up if any of these ops in here fail...
-
-	dir := cfg.IOFSAgentPath
-	if dir == "" {
-		// /tmp should be a memory backed filesystem, where we can get user perms
-		// on the socket file (fdks must give write permissions to users on sock).
-		// /var/run is root only, hence this...
-		dir = "/tmp"
-	}
-
-	// create a tmpdir
-	iofsDir, err := ioutil.TempDir(dir, "iofs")
-	if err != nil {
-		return "", "", fmt.Errorf("cannot create tmpdir for iofs: %v", err)
-	}
-
-	opts := cfg.IOFSOpts
-	if opts == "" {
-		// opts = "size=1k,nr_inodes=8,mode=0777"
-	}
-	if cfg.IOFSAgentPath != "" && cfg.IOFSMountRoot != "" {
-		return iofsDir, filepath.Join(cfg.IOFSMountRoot, filepath.Base(iofsDir)), nil
-	}
-	return iofsDir, iofsDir, nil
-
-	// under tmpdir, create tmpfs
-	// TODO uh, yea, idk
-	//if cfg.IOFSAgentPath != "" {
-	//err = syscall.Mount("tmpfs", iofsDir, "tmpfs", uintptr( [>syscall.MS_NOEXEC|syscall.MS_NOSUID|syscall.MS_NODEV<] 0), opts)
-	//if err != nil {
-	//return "", fmt.Errorf("cannot mount/create tmpfs=%s", iofsDir)
-	//}
-	//}
-
-}
-
 func inotifyUDS(ctx context.Context, iofsDir string, awaitUDS chan<- error) {
 	// XXX(reed): I forgot how to plumb channels temporarily forgive me for this sin (inotify will timeout, this is just bad programming)
 	err := inotifyAwait(ctx, iofsDir)
@@ -1197,20 +1153,18 @@ func (a *agent) runHotReq(ctx context.Context, call *call, state ContainerState,
 // and stderr can be swapped out by new calls in the container.  input and
 // output must be copied in and out.
 type container struct {
-	id             string // contrived
+	id         string // contrived
-	image          string
+	image      string
-	env            map[string]string
+	env        map[string]string
-	extensions     map[string]string
+	extensions map[string]string
-	memory         uint64
+	memory     uint64
-	cpus           uint64
+	cpus       uint64
-	fsSize         uint64
+	fsSize     uint64
-	tmpFsSize      uint64
+	tmpFsSize  uint64
-	iofsAgentPath  string
+	iofs       iofs
-	iofsDockerPath string
+	timeout    time.Duration // cold only (superfluous, but in case)
-
+	logCfg     drivers.LoggerConfig
-	timeout time.Duration // cold only (superfluous, but in case)
+	close      func()
-	logCfg  drivers.LoggerConfig
-	close   func()
 
 	stdin  io.Reader
 	stdout io.Writer
@@ -1267,42 +1221,33 @@ func newHotContainer(ctx context.Context, call *call, cfg *Config) (*container,
 		stderr.Swap(newLineWriterWithBuffer(buf2, sec))
 	}
 
-	var iofsAgentPath, iofsDockerPath string
+	var iofs iofs
 	var err error
-	closer := func() {} // XXX(reed):
 	if call.Format == models.FormatHTTPStream {
 		// XXX(reed): we should also point stdout to stderr, and not have stdin
+		if cfg.IOFSEnableTmpfs {
+			iofs, err = newTmpfsIOFS(ctx, cfg)
+		} else {
+			iofs, err = newDirectoryIOFS(ctx, cfg)
+		}
 
-		iofsAgentPath, iofsDockerPath, err = createIOFS(cfg)
 		if err != nil {
 			return nil, err
 		}
-
+	} else {
-		// XXX(reed): futz with this, we have to make sure shit gets cleaned up properly
+		iofs = &noopIOFS{}
-		closer = func() {
-			//err := syscall.Unmount(iofs, 0)
-			//if err != nil {
-			//common.Logger(ctx).WithError(err).Error("error unmounting iofs")
-			//}
-
-			err = os.RemoveAll(iofsAgentPath)
-			if err != nil {
-				common.Logger(ctx).WithError(err).Error("error removing iofs")
-			}
-		}
 	}
 
 	return &container{
-		id:             id, // XXX we could just let docker generate ids...
+		id:         id, // XXX we could just let docker generate ids...
-		image:          call.Image,
+		image:      call.Image,
-		env:            map[string]string(call.Config),
+		env:        map[string]string(call.Config),
-		extensions:     call.extensions,
+		extensions: call.extensions,
-		memory:         call.Memory,
+		memory:     call.Memory,
-		cpus:           uint64(call.CPUs),
+		cpus:       uint64(call.CPUs),
-		fsSize:         cfg.MaxFsSize,
+		fsSize:     cfg.MaxFsSize,
-		tmpFsSize:      uint64(call.TmpFsSize),
+		tmpFsSize:  uint64(call.TmpFsSize),
-		iofsAgentPath:  iofsAgentPath,
+		iofs:       iofs,
-		iofsDockerPath: iofsDockerPath,
 		logCfg: drivers.LoggerConfig{
 			URL: strings.TrimSpace(call.SyslogURL),
 			Tags: []drivers.LoggerTag{
@@ -1320,7 +1265,11 @@ func newHotContainer(ctx context.Context, call *call, cfg *Config) (*container,
 			for _, b := range bufs {
 				bufPool.Put(b)
 			}
-			closer() // XXX(reed): clean up
+			// iofs.Close MUST be called here or we will leak directories and/or tmpfs mounts!
+			if err = iofs.Close(); err != nil {
+				// Note: This is logged with the context of the container creation
+				common.Logger(ctx).WithError(err).Error("Error closing IOFS")
+			}
 		},
 	}, nil
 }
@@ -1362,8 +1311,8 @@ func (c *container) FsSize() uint64                     { return c.fsSize }
 func (c *container) TmpFsSize() uint64                  { return c.tmpFsSize }
 func (c *container) Extensions() map[string]string      { return c.extensions }
 func (c *container) LoggerConfig() drivers.LoggerConfig { return c.logCfg }
-func (c *container) UDSAgentPath() string               { return c.iofsAgentPath }
+func (c *container) UDSAgentPath() string               { return c.iofs.AgentPath() }
-func (c *container) UDSDockerPath() string              { return c.iofsDockerPath }
+func (c *container) UDSDockerPath() string              { return c.iofs.DockerPath() }
 
 // WriteStat publishes each metric in the specified Stats structure as a histogram metric
 func (c *container) WriteStat(ctx context.Context, stat drivers.Stat) {

api/agent/config.go

@@ -33,6 +33,7 @@ type Config struct {
 	DisableReadOnlyRootFs   bool          `json:"disable_readonly_rootfs"`
 	DisableTini             bool          `json:"disable_tini"`
 	DisableDebugUserLogs    bool          `json:"disable_debug_user_logs"`
+	IOFSEnableTmpfs         bool          `json:"iofs_enable_tmpfs"`
 	IOFSAgentPath           string        `json:"iofs_path"`
 	IOFSMountRoot           string        `json:"iofs_mount_root"`
 	IOFSOpts                string        `json:"iofs_opts"`
@@ -87,11 +88,12 @@ const (
 	// EnvDisableDebugUserLogs disables user function logs being logged at level debug. wise to enable for production.
 	EnvDisableDebugUserLogs = "FN_DISABLE_DEBUG_USER_LOGS"
 
+	// EnvIOFSEnableTmpfs enables creating a per-container tmpfs mount for the IOFS
+	EnvIOFSEnableTmpfs = "FN_IOFS_TMPFS"
 	// EnvIOFSPath is the path within fn server container of a directory to configure for unix socket files for each container
 	EnvIOFSPath = "FN_IOFS_PATH"
 	// EnvIOFSDockerPath determines the relative location on the docker host where iofs mounts should be prefixed with
 	EnvIOFSDockerPath = "FN_IOFS_DOCKER_PATH"
-
 	// EnvIOFSOpts are the options to set when mounting the iofs directory for unix socket files
 	EnvIOFSOpts = "FN_IOFS_OPTS"
 
@@ -137,21 +139,15 @@ func NewConfig() (*Config, error) {
 	err = setEnvStr(err, EnvIOFSPath, &cfg.IOFSAgentPath)
 	err = setEnvStr(err, EnvIOFSDockerPath, &cfg.IOFSMountRoot)
 	err = setEnvStr(err, EnvIOFSOpts, &cfg.IOFSOpts)
+	err = setEnvBool(err, EnvIOFSEnableTmpfs, &cfg.IOFSEnableTmpfs)
+	err = setEnvBool(err, EnvEnableNBResourceTracker, &cfg.EnableNBResourceTracker)
+	err = setEnvBool(err, EnvDisableReadOnlyRootFs, &cfg.DisableReadOnlyRootFs)
+	err = setEnvBool(err, EnvDisableDebugUserLogs, &cfg.DisableDebugUserLogs)
 
 	if err != nil {
 		return cfg, err
 	}
 
-	if _, ok := os.LookupEnv(EnvEnableNBResourceTracker); ok {
-		cfg.EnableNBResourceTracker = true
-	}
-	if _, ok := os.LookupEnv(EnvDisableReadOnlyRootFs); ok {
-		cfg.DisableReadOnlyRootFs = true
-	}
-	if _, ok := os.LookupEnv(EnvDisableDebugUserLogs); ok {
-		cfg.DisableDebugUserLogs = true
-	}
-
 	if cfg.EjectIdle == time.Duration(0) {
 		return cfg, fmt.Errorf("error %s cannot be zero", EnvEjectIdle)
 	}

api/agent/drivers/docker/cookie.go

@@ -105,7 +105,7 @@ func (c *cookie) configureTmpFs(log logrus.FieldLogger) {
 	c.opts.HostConfig.Tmpfs["/tmp"] = tmpFsOption
 }
 
-func (c *cookie) configureIOFs(log logrus.FieldLogger) {
+func (c *cookie) configureIOFS(log logrus.FieldLogger) {
 	path := c.task.UDSDockerPath()
 	if path == "" {
 		// TODO this should be required soon-ish

api/agent/drivers/docker/docker.go

@@ -242,7 +242,7 @@ func (drv *DockerDriver) CreateCookie(ctx context.Context, task drivers.Containe
 	cookie.configureTmpFs(log)
 	cookie.configureVolumes(log)
 	cookie.configureWorkDir(log)
-	cookie.configureIOFs(log)
+	cookie.configureIOFS(log)
 
 	// Order is important, if pool is enabled, it overrides pick network
 	drv.pickPool(ctx, cookie)

api/agent/iofs.go

@@ -0,0 +1,82 @@
+package agent
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+
+	"github.com/fnproject/fn/api/common"
+)
+
+type iofs interface {
+	io.Closer
+	AgentPath() string
+	DockerPath() string
+}
+
+type noopIOFS struct {
+}
+
+func (n *noopIOFS) AgentPath() string {
+	return ""
+}
+
+func (n *noopIOFS) DockerPath() string {
+	return ""
+}
+
+func (n *noopIOFS) Close() error {
+	return nil
+}
+
+type directoryIOFS struct {
+	agentPath  string
+	dockerPath string
+}
+
+func (d *directoryIOFS) AgentPath() string {
+	return d.agentPath
+}
+
+func (d *directoryIOFS) DockerPath() string {
+	return d.dockerPath
+}
+
+func (d *directoryIOFS) Close() error {
+	err := os.RemoveAll(d.agentPath)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func newDirectoryIOFS(ctx context.Context, cfg *Config) (*directoryIOFS, error) {
+	dir := cfg.IOFSAgentPath
+
+	// create a tmpdir
+	iofsAgentDir, err := ioutil.TempDir(dir, "iofs")
+	if err != nil {
+		if err := os.RemoveAll(iofsAgentDir); err != nil {
+			common.Logger(ctx).WithError(err).Error("failed to clean up iofs dir")
+		}
+		return nil, fmt.Errorf("cannot create tmpdir for iofs: %v", err)
+	}
+
+	if cfg.IOFSMountRoot != "" {
+		iofsRelPath, err := filepath.Rel(dir, iofsAgentDir)
+		if err != nil {
+			if err := os.RemoveAll(iofsAgentDir); err != nil {
+				common.Logger(ctx).WithError(err).Error("failed to clean up iofs dir")
+			}
+			return nil, fmt.Errorf("cannot relativise iofs path: %v", err)
+		}
+		iofsDockerDir := filepath.Join(cfg.IOFSMountRoot, iofsRelPath)
+		return &directoryIOFS{iofsAgentDir, iofsDockerDir}, nil
+	}
+	return &directoryIOFS{iofsAgentDir, iofsAgentDir}, nil
+}
+
+var _ iofs = &directoryIOFS{}

api/agent/iofs_linux.go

@@ -0,0 +1,38 @@
+package agent
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/fnproject/fn/api/common"
+	"golang.org/x/sys/unix"
+)
+
+type tmpfsIOFS struct {
+	directoryIOFS
+}
+
+func (t *tmpfsIOFS) Close() error {
+	if err := unix.Unmount(t.AgentPath(), 0); err != nil {
+		// At this point we don't have a lot of choice but to leak the directory and mount
+		return err
+	}
+	return t.directoryIOFS.Close()
+}
+
+func newTmpfsIOFS(ctx context.Context, cfg *Config) (*tmpfsIOFS, error) {
+	dirIOFS, err := newDirectoryIOFS(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+	if err = unix.Mount("tmpfs", dirIOFS.AgentPath(), "tmpfs", uintptr(unix.MS_NOEXEC|unix.MS_NOSUID|unix.MS_NODEV), cfg.IOFSOpts); err != nil {
+		// Best effort to clean up after failure. If the dirIOFS.Close() fails we're not going to see the error though...
+		if err := dirIOFS.Close(); err != nil {
+			common.Logger(ctx).WithError(err).Error("failed to cleanup iofs dir")
+		}
+		return nil, fmt.Errorf("cannot mount/create tmpfs at %s", dirIOFS.AgentPath())
+	}
+	return &tmpfsIOFS{*dirIOFS}, nil
+}
+
+var _ iofs = &tmpfsIOFS{}

api/agent/iofs_notlinux.go

@@ -0,0 +1,22 @@
+// +build !linux
+
+package agent
+
+import (
+	"context"
+	"errors"
+)
+
+type tmpfsIOFS struct {
+	directoryIOFS
+}
+
+func (t *tmpfsIOFS) Close() error {
+	return t.directoryIOFS.Close()
+}
+
+func newTmpfsIOFS(ctx context.Context, cfg *Config) (*tmpfsIOFS, error) {
+	return nil, errors.New("tmpfs IOFS not supported on macOS")
+}
+
+var _ iofs = &tmpfsIOFS{}