Merge pull request #133 from sboeuf/fix_device_passing

grpc: Handle hotplugged devices

device.go

@@ -0,0 +1,157 @@
+//
+// Copyright (c) 2018 Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/kata-containers/agent/pkg/uevent"
+	pb "github.com/kata-containers/agent/protocols/grpc"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
+	grpcStatus "google.golang.org/grpc/status"
+)
+
+type deviceDriversHandler func(device pb.Device, spec *pb.Spec) error
+
+var deviceDriversHandlerList = map[string]deviceDriversHandler{
+	"blk": blockDeviceHandler,
+}
+
+func blockDeviceHandler(device pb.Device, spec *pb.Spec) error {
+	// First need to make sure the expected device shows up properly,
+	// and then we need to retrieve its device info (such as major and
+	// minor numbers), useful to update the device provided
+	// through the OCI specification.
+	if err := waitForDevice(device.VmPath); err != nil {
+		return err
+	}
+
+	// If no ContainerPath is provided, this means we don't expect the
+	// device to be updated in the specification. We can return from here.
+	if device.ContainerPath == "" {
+		return nil
+	}
+
+	// At this point in the code, we assume the specification will be
+	// updated, meaning we should make sure we have valid pointers here.
+	if spec.Linux == nil || len(spec.Linux.Devices) == 0 {
+		return grpcStatus.Errorf(codes.Internal,
+			"No devices found from the spec, cannot update")
+	}
+
+	stat := syscall.Stat_t{}
+	if err := syscall.Stat(device.VmPath, &stat); err != nil {
+		return err
+	}
+
+	major := int64(stat.Rdev / 256)
+	minor := int64(stat.Rdev % 256)
+
+	agentLog.WithFields(logrus.Fields{
+		"device-path":  device.VmPath,
+		"device-major": major,
+		"device-minor": minor,
+	}).Info("handling block device")
+
+	// Update the spec
+	updated := false
+	for idx, d := range spec.Linux.Devices {
+		if d.Path == device.ContainerPath {
+			agentLog.WithFields(logrus.Fields{
+				"device-path":        device.VmPath,
+				"host-device-major":  spec.Linux.Devices[idx].Major,
+				"host-device-minor":  spec.Linux.Devices[idx].Minor,
+				"guest-device-major": major,
+				"guest-device-minor": minor,
+			}).Info("updating block device major/minor into the spec")
+			spec.Linux.Devices[idx].Major = major
+			spec.Linux.Devices[idx].Minor = minor
+			updated = true
+			break
+		}
+	}
+
+	if !updated {
+		return grpcStatus.Errorf(codes.Internal,
+			"Should have found a matching device %s in the spec",
+			device.VmPath)
+	}
+
+	return nil
+}
+
+func waitForDevice(devicePath string) error {
+	deviceName := strings.TrimPrefix(devicePath, devPrefix)
+
+	if _, err := os.Stat(devicePath); err == nil {
+		return nil
+	}
+
+	uEvHandler, err := uevent.NewHandler()
+	if err != nil {
+		return err
+	}
+	defer uEvHandler.Close()
+
+	fieldLogger := agentLog.WithField("device", deviceName)
+
+	// Check if the device already exists.
+	if _, err := os.Stat(devicePath); err == nil {
+		fieldLogger.Info("Device already hotplugged, quit listening")
+		return nil
+	}
+
+	fieldLogger.Info("Started listening for uevents for device hotplug")
+
+	// Channel to signal when desired uevent has been received.
+	done := make(chan bool)
+
+	go func() {
+		// This loop will be either ended if the hotplugged device is
+		// found by listening to the netlink socket, or it will end
+		// after the function returns and the uevent handler is closed.
+		for {
+			uEv, err := uEvHandler.Read()
+			if err != nil {
+				fieldLogger.Error(err)
+				continue
+			}
+
+			fieldLogger = fieldLogger.WithFields(logrus.Fields{
+				"uevent-action":    uEv.Action,
+				"uevent-devpath":   uEv.DevPath,
+				"uevent-subsystem": uEv.SubSystem,
+				"uevent-seqnum":    uEv.SeqNum,
+			})
+
+			fieldLogger.Info("Got uevent")
+
+			if uEv.Action == "add" &&
+				filepath.Base(uEv.DevPath) == deviceName {
+				fieldLogger.Info("Hotplug event received")
+				break
+			}
+		}
+
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Duration(timeoutHotplug) * time.Second):
+		return grpcStatus.Errorf(codes.DeadlineExceeded,
+			"Timeout reached after %ds waiting for device %s",
+			timeoutHotplug, deviceName)
+	}
+
+	return nil
+}

device_test.go

@@ -0,0 +1,70 @@
+//
+// Copyright (c) 2018 Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+package main
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+
+	pb "github.com/kata-containers/agent/protocols/grpc"
+	"github.com/stretchr/testify/assert"
+)
+
+func createFakeDevicePath() (string, error) {
+	f, err := ioutil.TempFile("", "fake-dev-path")
+	if err != nil {
+		return "", err
+	}
+	path := f.Name()
+	f.Close()
+
+	return path, nil
+}
+
+func testblockDeviceHandlerSuccessful(t *testing.T, device pb.Device, spec *pb.Spec) {
+	devPath, err := createFakeDevicePath()
+	assert.Nil(t, err, "Fake device path creation failed: %v", err)
+	defer os.RemoveAll(devPath)
+
+	device.VmPath = devPath
+
+	err = blockDeviceHandler(device, spec)
+	assert.Nil(t, err, "blockDeviceHandler() failed: %v", err)
+}
+
+func TestBlockDeviceHandlerNilLinuxSpecSuccessful(t *testing.T) {
+	spec := &pb.Spec{}
+
+	testblockDeviceHandlerSuccessful(t, pb.Device{}, spec)
+}
+
+func testblockDeviceHandlerFailure(t *testing.T, device pb.Device, spec *pb.Spec) {
+	devPath, err := createFakeDevicePath()
+	assert.Nil(t, err, "Fake device path creation failed: %v", err)
+	defer os.RemoveAll(devPath)
+
+	device.VmPath = devPath
+	device.ContainerPath = "some-not-empty-path"
+
+	err = blockDeviceHandler(device, spec)
+	assert.NotNil(t, err, "blockDeviceHandler() should have failed")
+}
+
+func TestBlockDeviceHandlerNilLinuxSpecFailure(t *testing.T) {
+	spec := &pb.Spec{}
+
+	testblockDeviceHandlerFailure(t, pb.Device{}, spec)
+}
+
+func TestBlockDeviceHandlerEmptyLinuxDevicesSpecFailure(t *testing.T) {
+	spec := &pb.Spec{
+		Linux: &pb.Linux{},
+	}
+
+	testblockDeviceHandlerFailure(t, pb.Device{}, spec)
+}

grpc.go

@@ -323,6 +323,26 @@ func (a *agentGRPC) updateContainerConfig(spec *specs.Spec, config *configs.Conf
 	return a.updateContainerConfigPrivileges(spec, config)
 }
 
+func addDevices(devices []*pb.Device, spec *pb.Spec) error {
+	for _, device := range devices {
+		if device == nil {
+			continue
+		}
+
+		devHandler, ok := deviceDriversHandlerList[device.Type]
+		if !ok {
+			return grpcStatus.Errorf(codes.InvalidArgument,
+				"Unknown device type %q", device.Type)
+		}
+
+		if err := devHandler(*device, spec); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (a *agentGRPC) CreateContainer(ctx context.Context, req *pb.CreateContainerRequest) (*gpb.Empty, error) {
 	if a.sandbox.running == false {
 		return emptyResp, grpcStatus.Error(codes.FailedPrecondition, "Sandbox not started, impossible to run a new container")
@@ -338,6 +358,10 @@ func (a *agentGRPC) CreateContainer(ctx context.Context, req *pb.CreateContainer
 		agentLog.WithError(err).Warn("Could not rescan PCI bus")
 	}
 
+	if err := addDevices(req.Devices, req.OCI); err != nil {
+		return emptyResp, err
+	}
+
 	mountList, err := addMounts(req.Storages)
 	if err != nil {
 		return emptyResp, err

grpc_test.go

@@ -7,9 +7,11 @@
 package main
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
+	pb "github.com/kata-containers/agent/protocols/grpc"
 	"github.com/opencontainers/runc/libcontainer/configs"
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/stretchr/testify/assert"
@@ -102,3 +104,86 @@ func TestUpdateContainerConfigPrivilegesNoNewPrivileges(t *testing.T) {
 		testUpdateContainerConfigPrivileges(t, spec, config, expectedConfig)
 	}
 }
+
+func testAddDevicesSuccessful(t *testing.T, devices []*pb.Device, spec *pb.Spec) {
+	err := addDevices(devices, spec)
+	assert.Nil(t, err, "addDevices() failed: %v", err)
+}
+
+func TestAddDevicesEmptyDevicesSuccessful(t *testing.T) {
+	var devices []*pb.Device
+	spec := &pb.Spec{}
+
+	testAddDevicesSuccessful(t, devices, spec)
+}
+
+func TestAddDevicesNilMountsSuccessful(t *testing.T) {
+	devices := []*pb.Device{
+		nil,
+	}
+
+	spec := &pb.Spec{}
+
+	testAddDevicesSuccessful(t, devices, spec)
+}
+
+func noopDeviceHandlerReturnNil(device pb.Device, spec *pb.Spec) error {
+	return nil
+}
+
+func noopDeviceHandlerReturnError(device pb.Device, spec *pb.Spec) error {
+	return fmt.Errorf("Noop handler failure")
+}
+
+func TestAddDevicesNoopHandlerSuccessful(t *testing.T) {
+	noopHandlerTag := "noop"
+	deviceDriversHandlerList = map[string]deviceDriversHandler{
+		noopHandlerTag: noopDeviceHandlerReturnNil,
+	}
+
+	devices := []*pb.Device{
+		{
+			Type: noopHandlerTag,
+		},
+	}
+
+	spec := &pb.Spec{}
+
+	testAddDevicesSuccessful(t, devices, spec)
+}
+
+func testAddDevicesFailure(t *testing.T, devices []*pb.Device, spec *pb.Spec) {
+	err := addDevices(devices, spec)
+	assert.NotNil(t, err, "addDevices() should have failed")
+}
+
+func TestAddDevicesUnknownHandlerFailure(t *testing.T) {
+	deviceDriversHandlerList = map[string]deviceDriversHandler{}
+
+	devices := []*pb.Device{
+		{
+			Type: "unknown",
+		},
+	}
+
+	spec := &pb.Spec{}
+
+	testAddDevicesFailure(t, devices, spec)
+}
+
+func TestAddDevicesNoopHandlerFailure(t *testing.T) {
+	noopHandlerTag := "noop"
+	deviceDriversHandlerList = map[string]deviceDriversHandler{
+		noopHandlerTag: noopDeviceHandlerReturnError,
+	}
+
+	devices := []*pb.Device{
+		{
+			Type: noopHandlerTag,
+		},
+	}
+
+	spec := &pb.Spec{}
+
+	testAddDevicesFailure(t, devices, spec)
+}