Initial support for nvme

protokube/pkg/protokube/aws_volume.go

@@ -19,21 +19,22 @@ package protokube
 import (
 	"fmt"
 	"net"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
 
-	"k8s.io/kops/protokube/pkg/gossip"
-	gossipaws "k8s.io/kops/protokube/pkg/gossip/aws"
-	"k8s.io/kops/upup/pkg/fi/cloudup/awsup"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/ec2metadata"
 	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/golang/glog"
 	"k8s.io/kops/protokube/pkg/etcd"
+	"k8s.io/kops/protokube/pkg/gossip"
+	gossipaws "k8s.io/kops/protokube/pkg/gossip/aws"
+	"k8s.io/kops/upup/pkg/fi/cloudup/awsup"
 )
 
 var devices = []string{"/dev/xvdu", "/dev/xvdv", "/dev/xvdx", "/dev/xvdx", "/dev/xvdy", "/dev/xvdz"}
@@ -277,6 +278,72 @@ func (a *AWSVolumes) FindVolumes() ([]*Volume, error) {
 	return a.findVolumes(request)
 }
 
+// FindMountedVolume implements Volumes::FindMountedVolume
+func (v *AWSVolumes) FindMountedVolume(volume *Volume) (string, error) {
+	device := volume.LocalDevice
+
+	_, err := os.Stat(pathFor(device))
+	if err == nil {
+		return device, nil
+	}
+	if !os.IsNotExist(err) {
+		return "", fmt.Errorf("error checking for device %q: %v", device, err)
+	}
+
+	if volume.ID != "" {
+		expected := volume.ID
+		expected = "nvme-Amazon_Elastic_Block_Store_" + strings.Replace(expected, "-", "", -1)
+
+		// Look for nvme devices
+		// On AWS, nvme volumes are not mounted on a device path, but are instead mounted on an nvme device
+		// We must identify the correct volume by matching the nvme info
+		device, err := findNvmeVolume(expected)
+		if err != nil {
+			return "", fmt.Errorf("error checking for nvme volume %q: %v", expected, err)
+		}
+		if device != "" {
+			glog.Infof("found nvme volume %q at %q", expected, device)
+			return device, nil
+		}
+	}
+
+	return "", nil
+}
+
+func findNvmeVolume(findName string) (device string, err error) {
+	p := pathFor("/dev/disk/by-id/" + findName)
+	stat, err := os.Lstat(p)
+	if err != nil {
+		if os.IsNotExist(err) {
+			glog.V(4).Infof("nvme path not found %q", p)
+			return "", nil
+		}
+		return "", fmt.Errorf("error getting stat of %q: %v", p, err)
+	}
+
+	if stat.Mode()&os.ModeSymlink != os.ModeSymlink {
+		glog.Warningf("nvme file %q found, but was not a symlink", p)
+		return "", nil
+	}
+
+	resolved, err := filepath.EvalSymlinks(p)
+	if err != nil {
+		return "", fmt.Errorf("error reading target of symlink %q: %v", p, err)
+	}
+
+	// Reverse pathFor
+	devPath := pathFor("/dev")
+	if strings.HasPrefix(resolved, devPath) {
+		resolved = strings.Replace(resolved, devPath, "/dev", 1)
+	}
+
+	if !strings.HasPrefix(resolved, "/dev") {
+		return "", fmt.Errorf("resolved symlink for %q was unexpected: %q", p, resolved)
+	}
+
+	return resolved, nil
+}
+
 // assignDevice picks a hopefully unused device and reserves it for the volume attachment
 func (a *AWSVolumes) assignDevice(volumeID string) (string, error) {
 	a.mutex.Lock()
@@ -364,7 +431,7 @@ func (a *AWSVolumes) AttachVolume(volume *Volume) error {
 		switch v.Status {
 		case "attaching":
 			glog.V(2).Infof("Waiting for volume %q to be attached (currently %q)", volumeID, v.Status)
-		// continue looping
+			// continue looping
 
 		default:
 			return fmt.Errorf("Observed unexpected volume state %q", v.Status)

protokube/pkg/protokube/gce_volume.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kops/protokube/pkg/gossip"
 	gossipgce "k8s.io/kops/protokube/pkg/gossip/gce"
 	"k8s.io/kops/upup/pkg/fi/cloudup/gce"
+	"os"
 )
 
 // GCEVolumes is the Volumes implementation for GCE
@@ -76,7 +77,7 @@ func (a *GCEVolumes) ClusterID() string {
 	return a.clusterName
 }
 
-// ClusterID returns the current GCE project
+// Project returns the current GCE project
 func (a *GCEVolumes) Project() string {
 	return a.project
 }
@@ -310,6 +311,20 @@ func (v *GCEVolumes) FindVolumes() ([]*Volume, error) {
 	return volumes, nil
 }
 
+// FindMountedVolume implements Volumes::FindMountedVolume
+func (v *GCEVolumes) FindMountedVolume(volume *Volume) (string, error) {
+	device := volume.LocalDevice
+
+	_, err := os.Stat(pathFor(device))
+	if err == nil {
+		return device, nil
+	}
+	if os.IsNotExist(err) {
+		return "", nil
+	}
+	return "", fmt.Errorf("error checking for device %q: %v", device, err)
+}
+
 // AttachVolume attaches the specified volume to this instance, returning the mountpoint & nil if successful
 func (v *GCEVolumes) AttachVolume(volume *Volume) error {
 	volumeName := volume.ID

protokube/pkg/protokube/volume_mounter.go

@@ -71,7 +71,7 @@ func (k *VolumeMountController) mountMasterVolumes() ([]*Volume, error) {
 
 		glog.Infof("Doing safe-format-and-mount of %s to %s", v.LocalDevice, mountpoint)
 		fstype := ""
-		err = k.safeFormatAndMount(v.LocalDevice, mountpoint, fstype)
+		err = k.safeFormatAndMount(v, mountpoint, fstype)
 		if err != nil {
 			glog.Warningf("unable to mount master volume: %q", err)
 			continue
@@ -90,20 +90,24 @@ func (k *VolumeMountController) mountMasterVolumes() ([]*Volume, error) {
 	return volumes, nil
 }
 
-func (k *VolumeMountController) safeFormatAndMount(device string, mountpoint string, fstype string) error {
+func (k *VolumeMountController) safeFormatAndMount(volume *Volume, mountpoint string, fstype string) error {
 	// Wait for the device to show up
+	device := ""
 	for {
-		_, err := os.Stat(pathFor(device))
-		if err == nil {
-			break
+		found, err := k.provider.FindMountedVolume(volume)
+		if err != nil {
+			return err
 		}
-		if !os.IsNotExist(err) {
-			return fmt.Errorf("error checking for device %q: %v", device, err)
+
+		if found != "" {
+			device = found
+			break
 		}
-		glog.Infof("Waiting for device %q to be attached", device)
+
+		glog.Infof("Waiting for volume %q to be attached", volume.ID)
 		time.Sleep(1 * time.Second)
 	}
-	glog.Infof("Found device %q", device)
+	glog.Infof("Found volume %q mounted at device %q", volume.ID, device)
 
 	safeFormatAndMount := &mount.SafeFormatAndMount{}
 

protokube/pkg/protokube/volumes.go

@@ -23,13 +23,18 @@ import (
 type Volumes interface {
 	AttachVolume(volume *Volume) error
 	FindVolumes() ([]*Volume, error)
+
+	// FindMountedVolume returns the device (e.g. /dev/sda) where the volume is mounted
+	// If not found, it returns "", nil
+	// On error, it returns "", err
+	FindMountedVolume(volume *Volume) (device string, err error)
 }
 
 type Volume struct {
 	// ID is the cloud-provider identifier for the volume
 	ID string
 
-	// Device is set if the volume is attached to the local machine
+	// LocalDevice is set if the volume is attached to the local machine
 	LocalDevice string
 
 	// AttachedTo is set to the ID of the machine the volume is attached to, or "" if not attached

protokube/pkg/protokube/vsphere_volume.go

@@ -30,6 +30,7 @@ import (
 	"github.com/golang/glog"
 	etcdmanager "k8s.io/kops/protokube/pkg/etcd"
 	"k8s.io/kops/upup/pkg/fi/cloudup/vsphere"
+	"os"
 )
 
 const VolumeMetaDataFile = "/vol-metadata/metadata.json"
@@ -97,6 +98,20 @@ func (v *VSphereVolumes) FindVolumes() ([]*Volume, error) {
 	return volumes, nil
 }
 
+// FindMountedVolume implements Volumes::FindMountedVolume
+func (v *VSphereVolumes) FindMountedVolume(volume *Volume) (string, error) {
+	device := volume.LocalDevice
+
+	_, err := os.Stat(pathFor(device))
+	if err == nil {
+		return device, nil
+	}
+	if os.IsNotExist(err) {
+		return "", nil
+	}
+	return "", fmt.Errorf("error checking for device %q: %v", device, err)
+}
+
 func getDevice(mountPoint string) (string, error) {
 	if runtime.GOOS == "linux" {
 		cmd := "lsblk"