Merge pull request #1994 from stefanha/virtio-fs-v0.3

Virtio-fs v0.3 support

virtcontainers/kata_agent.go

@@ -59,7 +59,7 @@ var (
 	errorMissingOCISpec         = errors.New("Missing OCI specification")
 	defaultKataHostSharedDir    = "/run/kata-containers/shared/sandboxes/"
 	defaultKataGuestSharedDir   = "/run/kata-containers/shared/containers/"
-	mountGuest9pTag             = "kataShared"
+	mountGuestTag               = "kataShared"
 	defaultKataGuestSandboxDir  = "/run/kata-containers/sandbox/"
 	type9pFs                    = "9p"
 	typeVirtioFS                = "virtio_fs"
@@ -72,7 +72,7 @@ var (
 	kataNvdimmDevType           = "nvdimm"
 	kataVirtioFSDevType         = "virtio-fs"
 	sharedDir9pOptions          = []string{"trans=virtio,version=9p2000.L,cache=mmap", "nodev"}
-	sharedDirVirtioFSOptions    = []string{"default_permissions,allow_other,rootmode=040000,user_id=0,group_id=0,dax,tag=" + mountGuest9pTag, "nodev"}
+	sharedDirVirtioFSOptions    = []string{"default_permissions,allow_other,rootmode=040000,user_id=0,group_id=0", "nodev"}
 	sharedDirVirtioFSDaxOptions = "dax"
 	shmDir                      = "shm"
 	kataEphemeralDevType        = "ephemeral"
@@ -401,7 +401,7 @@ func (k *kataAgent) configure(h hypervisor, id, sharePath string, builtin bool,
 	// Create shared directory and add the shared volume if filesystem sharing is supported.
 	// This volume contains all bind mounted container bundles.
 	sharedVolume := types.Volume{
-		MountTag: mountGuest9pTag,
+		MountTag: mountGuestTag,
 		HostPath: sharePath,
 	}
 
@@ -872,7 +872,7 @@ func setupStorages(sandbox *Sandbox) []*grpc.Storage {
 			}
 			sharedVolume := &grpc.Storage{
 				Driver:     kataVirtioFSDevType,
-				Source:     "none",
+				Source:     mountGuestTag,
 				MountPoint: kataGuestSharedDir(),
 				Fstype:     typeVirtioFS,
 				Options:    sharedDirVirtioFSOptions,
@@ -884,7 +884,7 @@ func setupStorages(sandbox *Sandbox) []*grpc.Storage {
 
 			sharedVolume := &grpc.Storage{
 				Driver:     kata9pDevType,
-				Source:     mountGuest9pTag,
+				Source:     mountGuestTag,
 				MountPoint: kataGuestSharedDir(),
 				Fstype:     type9pFs,
 				Options:    sharedDir9pOptions,

virtcontainers/qemu.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"math"
+	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -602,15 +603,16 @@ func (q *qemu) vhostFSSocketPath(id string) (string, error) {
 	return utils.BuildSocketPath(store.RunVMStoragePath(), id, vhostFSSocket)
 }
 
-func (q *qemu) virtiofsdArgs(sockPath string) []string {
+func (q *qemu) virtiofsdArgs(fd uintptr) []string {
 	// The daemon will terminate when the vhost-user socket
 	// connection with QEMU closes.  Therefore we do not keep track
 	// of this child process after returning from this function.
 	sourcePath := filepath.Join(kataHostSharedDir(), q.id)
 	args := []string{
-		"-o", "vhost_user_socket=" + sockPath,
+		fmt.Sprintf("--fd=%v", fd),
 		"-o", "source=" + sourcePath,
-		"-o", "cache=" + q.config.VirtioFSCache}
+		"-o", "cache=" + q.config.VirtioFSCache,
+		"--syslog", "-o", "no_posix_lock"}
 	if q.config.Debug {
 		args = append(args, "-d")
 	} else {
@@ -623,82 +625,41 @@ func (q *qemu) virtiofsdArgs(sockPath string) []string {
 	return args
 }
 
-func (q *qemu) setupVirtiofsd(timeout int) (remain int, err error) {
+func (q *qemu) setupVirtiofsd() (err error) {
+	var listener *net.UnixListener
+	var fd *os.File
+
 	sockPath, err := q.vhostFSSocketPath(q.id)
 	if err != nil {
-		return 0, err
+		return err
 	}
 
-	cmd := exec.Command(q.config.VirtioFSDaemon, q.virtiofsdArgs(sockPath)...)
-	stderr, err := cmd.StderrPipe()
+	listener, err = net.ListenUnix("unix", &net.UnixAddr{
+		Name: sockPath,
+		Net:  "unix",
+	})
 	if err != nil {
-		return 0, err
+		return err
 	}
+	listener.SetUnlinkOnClose(false)
 
-	if err = cmd.Start(); err != nil {
-		return 0, err
-	}
-	defer func() {
-		if err != nil {
-			cmd.Process.Kill()
-		} else {
-			q.state.VirtiofsdPid = cmd.Process.Pid
-		}
-	}()
-
-	// Wait for socket to become available
-	sockReady := make(chan error, 1)
-	timeStart := time.Now()
-	go func() {
-		scanner := bufio.NewScanner(stderr)
-		var sent bool
-		for scanner.Scan() {
-			if q.config.Debug {
-				q.Logger().WithField("source", "virtiofsd").Debug(scanner.Text())
-			}
-			if !sent && strings.Contains(scanner.Text(), "Waiting for vhost-user socket connection...") {
-				sockReady <- nil
-				sent = true
-			}
-		}
-		if !sent {
-			if err := scanner.Err(); err != nil {
-				sockReady <- err
-			} else {
-				sockReady <- fmt.Errorf("virtiofsd did not announce socket connection")
-			}
-		}
-		q.Logger().Info("virtiofsd quits")
-		// Wait to release resources of virtiofsd process
-		cmd.Process.Wait()
-		q.stopSandbox()
-	}()
-
-	return q.waitVirtiofsd(timeStart, timeout, sockReady,
-		fmt.Sprintf("virtiofsd (pid=%d) socket %s", cmd.Process.Pid, sockPath))
-}
-
-func (q *qemu) waitVirtiofsd(start time.Time, timeout int, ready chan error, errMsg string) (int, error) {
-	var err error
-
-	timeoutDuration := time.Duration(timeout) * time.Second
-	select {
-	case err = <-ready:
-	case <-time.After(timeoutDuration):
-		err = fmt.Errorf("timed out waiting for %s", errMsg)
-	}
+	fd, err = listener.File()
+	listener.Close() // no longer needed since fd is a dup
+	listener = nil
 	if err != nil {
-		return 0, err
+		return err
 	}
 
-	// Now reduce timeout by the elapsed time
-	elapsed := time.Since(start)
-	if elapsed < timeoutDuration {
-		timeout = timeout - int(elapsed.Seconds())
-	} else {
-		timeout = 0
+	const sockFd = 3 // Cmd.ExtraFiles[] fds are numbered starting from 3
+	cmd := exec.Command(q.config.VirtioFSDaemon, q.virtiofsdArgs(sockFd)...)
+	cmd.ExtraFiles = append(cmd.ExtraFiles, fd)
+
+	err = cmd.Start()
+	if err == nil {
+		q.state.VirtiofsdPid = cmd.Process.Pid
 	}
-	return timeout, nil
+	fd.Close()
+	return err
 }
 
 // startSandbox will start the Sandbox's VM.
@@ -746,7 +707,7 @@ func (q *qemu) startSandbox(timeout int) error {
 	}()
 
 	if q.config.SharedFS == config.VirtioFS {
-		timeout, err = q.setupVirtiofsd(timeout)
+		err = q.setupVirtiofsd()
 		if err != nil {
 			return err
 		}
@@ -769,7 +730,7 @@ func (q *qemu) startSandbox(timeout int) error {
 		return fmt.Errorf("failed to launch qemu: %s, error messages from qemu log: %s", err, strErr)
 	}
 
-	err = q.waitSandbox(timeout) // the virtiofsd deferred checks err's value
+	err = q.waitSandbox(timeout)
 	if err != nil {
 		return err
 	}

virtcontainers/qemu_test.go

@@ -13,7 +13,6 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
-	"time"
 
 	govmmQemu "github.com/intel/govmm/qemu"
 	"github.com/kata-containers/runtime/virtcontainers/device/config"
@@ -503,36 +502,16 @@ func TestQemuVirtiofsdArgs(t *testing.T) {
 		kataHostSharedDir = savedKataHostSharedDir
 	}()
 
-	result := "-o vhost_user_socket=bar1 -o source=test-share-dir/foo -o cache=none -d"
-	args := q.virtiofsdArgs("bar1")
+	result := "--fd=123 -o source=test-share-dir/foo -o cache=none --syslog -o no_posix_lock -d"
+	args := q.virtiofsdArgs(123)
 	assert.Equal(strings.Join(args, " "), result)
 
 	q.config.Debug = false
-	result = "-o vhost_user_socket=bar2 -o source=test-share-dir/foo -o cache=none -f"
-	args = q.virtiofsdArgs("bar2")
+	result = "--fd=123 -o source=test-share-dir/foo -o cache=none --syslog -o no_posix_lock -f"
+	args = q.virtiofsdArgs(123)
 	assert.Equal(strings.Join(args, " "), result)
 }
 
-func TestQemuWaitVirtiofsd(t *testing.T) {
-	assert := assert.New(t)
-
-	q := &qemu{}
-
-	ready := make(chan error, 1)
-	timeout := 5
-
-	ready <- nil
-	remain, err := q.waitVirtiofsd(time.Now(), timeout, ready, "")
-	assert.Nil(err)
-	assert.True(remain <= timeout)
-	assert.True(remain >= 0)
-
-	timeout = 0
-	remain, err = q.waitVirtiofsd(time.Now(), timeout, ready, "")
-	assert.NotNil(err)
-	assert.True(remain == 0)
-}
-
 func TestQemuGetpids(t *testing.T) {
 	assert := assert.New(t)