Merge branch 'master' into fluentd-1.1.0

monotek · web-flow · commit d4ba0ab3e0e3 · 2018-01-19T17:29:49.000+01:00
diff --git a/pkg/cloudprovider/providers/openstack/BUILD b/pkg/cloudprovider/providers/openstack/BUILD
@@ -26,6 +26,7 @@ go_library(
         "//pkg/controller:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/github.com/gophercloud/gophercloud:go_default_library",
         "//vendor/github.com/gophercloud/gophercloud/openstack:go_default_library",
diff --git a/pkg/cloudprovider/providers/openstack/openstack.go b/pkg/cloudprovider/providers/openstack/openstack.go
@@ -319,6 +319,22 @@ func mapNodeNameToServerName(nodeName types.NodeName) string {
 	return string(nodeName)
 }
 
+// getNodeNameByID maps instanceid to types.NodeName
+func (os *OpenStack) GetNodeNameByID(instanceID string) (types.NodeName, error) {
+	client, err := os.NewComputeV2()
+	var nodeName types.NodeName
+	if err != nil {
+		return nodeName, err
+	}
+
+	server, err := servers.Get(client, instanceID).Extract()
+	if err != nil {
+		return nodeName, err
+	}
+	nodeName = mapServerToNodeName(server)
+	return nodeName, nil
+}
+
 // mapServerToNodeName maps an OpenStack Server to a k8s NodeName
 func mapServerToNodeName(server *servers.Server) types.NodeName {
 	// Node names are always lowercase, and (at least)
@@ -346,11 +362,14 @@ func foreachServer(client *gophercloud.ServiceClient, opts servers.ListOptsBuild
 	return err
 }
 
-func getServerByName(client *gophercloud.ServiceClient, name types.NodeName) (*servers.Server, error) {
+func getServerByName(client *gophercloud.ServiceClient, name types.NodeName, showOnlyActive bool) (*servers.Server, error) {
 	opts := servers.ListOpts{
-		Name:   fmt.Sprintf("^%s$", regexp.QuoteMeta(mapNodeNameToServerName(name))),
-		Status: "ACTIVE",
+		Name: fmt.Sprintf("^%s$", regexp.QuoteMeta(mapNodeNameToServerName(name))),
 	}
+	if showOnlyActive {
+		opts.Status = "ACTIVE"
+	}
+
 	pager := servers.List(client, opts)
 
 	serverList := make([]servers.Server, 0, 1)
@@ -432,7 +451,7 @@ func nodeAddresses(srv *servers.Server) ([]v1.NodeAddress, error) {
 }
 
 func getAddressesByName(client *gophercloud.ServiceClient, name types.NodeName) ([]v1.NodeAddress, error) {
-	srv, err := getServerByName(client, name)
+	srv, err := getServerByName(client, name, true)
 	if err != nil {
 		return nil, err
 	}
@@ -582,7 +601,7 @@ func (os *OpenStack) GetZoneByNodeName(nodeName types.NodeName) (cloudprovider.Z
 		return cloudprovider.Zone{}, err
 	}
 
-	srv, err := getServerByName(compute, nodeName)
+	srv, err := getServerByName(compute, nodeName, true)
 	if err != nil {
 		if err == ErrNotFound {
 			return cloudprovider.Zone{}, cloudprovider.InstanceNotFound
diff --git a/pkg/cloudprovider/providers/openstack/openstack_instances.go b/pkg/cloudprovider/providers/openstack/openstack_instances.go
@@ -43,7 +43,7 @@ func (os *OpenStack) Instances() (cloudprovider.Instances, bool) {
 		return nil, false
 	}
 
-	glog.V(1).Info("Claiming to support Instances")
+	glog.V(4).Info("Claiming to support Instances")
 
 	return &Instances{
 		compute: compute,
@@ -103,7 +103,7 @@ func (i *Instances) NodeAddressesByProviderID(providerID string) ([]v1.NodeAddre
 
 // ExternalID returns the cloud provider ID of the specified instance (deprecated).
 func (i *Instances) ExternalID(name types.NodeName) (string, error) {
-	srv, err := getServerByName(i.compute, name)
+	srv, err := getServerByName(i.compute, name, true)
 	if err != nil {
 		if err == ErrNotFound {
 			return "", cloudprovider.InstanceNotFound
@@ -151,7 +151,7 @@ func (os *OpenStack) InstanceID() (string, error) {
 
 // InstanceID returns the cloud provider ID of the specified instance.
 func (i *Instances) InstanceID(name types.NodeName) (string, error) {
-	srv, err := getServerByName(i.compute, name)
+	srv, err := getServerByName(i.compute, name, true)
 	if err != nil {
 		if err == ErrNotFound {
 			return "", cloudprovider.InstanceNotFound
@@ -184,7 +184,7 @@ func (i *Instances) InstanceTypeByProviderID(providerID string) (string, error)
 
 // InstanceType returns the type of the specified instance.
 func (i *Instances) InstanceType(name types.NodeName) (string, error) {
-	srv, err := getServerByName(i.compute, name)
+	srv, err := getServerByName(i.compute, name, true)
 
 	if err != nil {
 		return "", err
diff --git a/pkg/cloudprovider/providers/openstack/openstack_loadbalancer.go b/pkg/cloudprovider/providers/openstack/openstack_loadbalancer.go
@@ -537,7 +537,7 @@ func getSubnetIDForLB(compute *gophercloud.ServiceClient, node v1.Node) (string,
 	for _, intf := range interfaces {
 		for _, fixedIP := range intf.FixedIPs {
 			if fixedIP.IPAddress == ipAddress {
-				return intf.NetID, nil
+				return fixedIP.SubnetID, nil
 			}
 		}
 	}
@@ -551,7 +551,7 @@ func getNodeSecurityGroupIDForLB(compute *gophercloud.ServiceClient, nodes []*v1
 
 	for _, node := range nodes {
 		nodeName := types.NodeName(node.Name)
-		srv, err := getServerByName(compute, nodeName)
+		srv, err := getServerByName(compute, nodeName, true)
 		if err != nil {
 			return nodeSecurityGroupIDs.List(), err
 		}
diff --git a/pkg/cloudprovider/providers/openstack/openstack_routes.go b/pkg/cloudprovider/providers/openstack/openstack_routes.go
@@ -53,7 +53,7 @@ func (r *Routes) ListRoutes(clusterName string) ([]*cloudprovider.Route, error)
 	glog.V(4).Infof("ListRoutes(%v)", clusterName)
 
 	nodeNamesByAddr := make(map[string]types.NodeName)
-	err := foreachServer(r.compute, servers.ListOpts{Status: "ACTIVE"}, func(srv *servers.Server) (bool, error) {
+	err := foreachServer(r.compute, servers.ListOpts{}, func(srv *servers.Server) (bool, error) {
 		addrs, err := nodeAddresses(srv)
 		if err != nil {
 			return false, err
@@ -77,15 +77,11 @@ func (r *Routes) ListRoutes(clusterName string) ([]*cloudprovider.Route, error)
 
 	var routes []*cloudprovider.Route
 	for _, item := range router.Routes {
-		nodeName, ok := nodeNamesByAddr[item.NextHop]
-		if !ok {
-			// Not one of our routes?
-			glog.V(4).Infof("Skipping route with unknown nexthop %v", item.NextHop)
-			continue
-		}
+		nodeName, foundNode := nodeNamesByAddr[item.NextHop]
 		route := cloudprovider.Route{
 			Name:            item.DestinationCIDR,
-			TargetNode:      nodeName,
+			TargetNode:      nodeName, //empty if NextHop is unknown
+			Blackhole:       !foundNode,
 			DestinationCIDR: item.DestinationCIDR,
 		}
 		routes = append(routes, &route)
@@ -288,7 +284,7 @@ func (r *Routes) DeleteRoute(clusterName string, route *cloudprovider.Route) err
 }
 
 func getPortIDByIP(compute *gophercloud.ServiceClient, targetNode types.NodeName, ipAddress string) (string, error) {
-	srv, err := getServerByName(compute, targetNode)
+	srv, err := getServerByName(compute, targetNode, true)
 	if err != nil {
 		return "", err
 	}
diff --git a/pkg/cloudprovider/providers/openstack/openstack_volumes.go b/pkg/cloudprovider/providers/openstack/openstack_volumes.go
@@ -17,6 +17,7 @@ limitations under the License.
 package openstack
 
 import (
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"path"
@@ -25,7 +26,9 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/types"
 	k8s_volume "k8s.io/kubernetes/pkg/volume"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 
 	"github.com/gophercloud/gophercloud"
 	volumeexpand "github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/volumeactions"
@@ -318,7 +321,17 @@ func (os *OpenStack) AttachDisk(instanceID, volumeID string) (string, error) {
 			glog.V(4).Infof("Disk %s is already attached to instance %s", volumeID, instanceID)
 			return volume.ID, nil
 		}
-		return "", fmt.Errorf("disk %s is attached to a different instance (%s)", volumeID, volume.AttachedServerId)
+		nodeName, err := os.GetNodeNameByID(volume.AttachedServerId)
+		attachErr := fmt.Sprintf("disk %s path %s is attached to a different instance (%s)", volumeID, volume.AttachedDevice, volume.AttachedServerId)
+		if err != nil {
+			glog.Error(attachErr)
+			return "", errors.New(attachErr)
+		}
+		// using volume.AttachedDevice may cause problems because cinder does not report device path correctly see issue #33128
+		devicePath := volume.AttachedDevice
+		danglingErr := volumeutil.NewDanglingError(attachErr, nodeName, devicePath)
+		glog.V(2).Infof("Found dangling volume %s attached to node %s", volumeID, nodeName)
+		return "", danglingErr
 	}
 
 	startTime := time.Now()
@@ -578,6 +591,9 @@ func (os *OpenStack) GetAttachmentDiskPath(instanceID, volumeID string) (string,
 
 // DiskIsAttached queries if a volume is attached to a compute instance
 func (os *OpenStack) DiskIsAttached(instanceID, volumeID string) (bool, error) {
+	if instanceID == "" {
+		glog.Warningf("calling DiskIsAttached with empty instanceid: %s %s", instanceID, volumeID)
+	}
 	volume, err := os.getVolume(volumeID)
 	if err != nil {
 		return false, err
@@ -586,6 +602,29 @@ func (os *OpenStack) DiskIsAttached(instanceID, volumeID string) (bool, error) {
 	return instanceID == volume.AttachedServerId, nil
 }
 
+// DiskIsAttachedByName queries if a volume is attached to a compute instance by name
+func (os *OpenStack) DiskIsAttachedByName(nodeName types.NodeName, volumeID string) (bool, string, error) {
+	cClient, err := os.NewComputeV2()
+	if err != nil {
+		return false, "", err
+	}
+	srv, err := getServerByName(cClient, nodeName, false)
+	if err != nil {
+		if err == ErrNotFound {
+			// instance not found anymore in cloudprovider, assume that cinder is detached
+			return false, "", nil
+		} else {
+			return false, "", err
+		}
+	}
+	instanceID := "/" + srv.ID
+	if ind := strings.LastIndex(instanceID, "/"); ind >= 0 {
+		instanceID = instanceID[(ind + 1):]
+	}
+	attached, err := os.DiskIsAttached(instanceID, volumeID)
+	return attached, instanceID, err
+}
+
 // DisksAreAttached queries if a list of volumes are attached to a compute instance
 func (os *OpenStack) DisksAreAttached(instanceID string, volumeIDs []string) (map[string]bool, error) {
 	attached := make(map[string]bool)
@@ -600,6 +639,32 @@ func (os *OpenStack) DisksAreAttached(instanceID string, volumeIDs []string) (ma
 	return attached, nil
 }
 
+// DisksAreAttachedByName queries if a list of volumes are attached to a compute instance by name
+func (os *OpenStack) DisksAreAttachedByName(nodeName types.NodeName, volumeIDs []string) (map[string]bool, error) {
+	attached := make(map[string]bool)
+	cClient, err := os.NewComputeV2()
+	if err != nil {
+		return attached, err
+	}
+	srv, err := getServerByName(cClient, nodeName, false)
+	if err != nil {
+		if err == ErrNotFound {
+			// instance not found anymore, mark all volumes as detached
+			for _, volumeID := range volumeIDs {
+				attached[volumeID] = false
+			}
+			return attached, nil
+		} else {
+			return attached, err
+		}
+	}
+	instanceID := "/" + srv.ID
+	if ind := strings.LastIndex(instanceID, "/"); ind >= 0 {
+		instanceID = instanceID[(ind + 1):]
+	}
+	return os.DisksAreAttached(instanceID, volumeIDs)
+}
+
 // diskIsUsed returns true a disk is attached to any node.
 func (os *OpenStack) diskIsUsed(volumeID string) (bool, error) {
 	volume, err := os.getVolume(volumeID)
diff --git a/pkg/volume/cinder/attacher.go b/pkg/volume/cinder/attacher.go
@@ -27,7 +27,6 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -187,23 +186,7 @@ func (attacher *cinderDiskAttacher) VolumesAreAttached(specs []*volume.Spec, nod
 		volumeSpecMap[volumeSource.VolumeID] = spec
 	}
 
-	instanceID, err := attacher.nodeInstanceID(nodeName)
-	if err != nil {
-		if err == cloudprovider.InstanceNotFound {
-			// If node doesn't exist, OpenStack Nova will assume the volumes are not attached to it.
-			// Mark the volumes as detached and return false without error.
-			glog.Warningf("VolumesAreAttached: node %q does not exist.", nodeName)
-			for spec := range volumesAttachedCheck {
-				volumesAttachedCheck[spec] = false
-			}
-
-			return volumesAttachedCheck, nil
-		}
-
-		return volumesAttachedCheck, err
-	}
-
-	attachedResult, err := attacher.cinderProvider.DisksAreAttached(instanceID, volumeIDList)
+	attachedResult, err := attacher.cinderProvider.DisksAreAttachedByName(nodeName, volumeIDList)
 	if err != nil {
 		// Log error and continue with attach
 		glog.Errorf(
@@ -381,20 +364,10 @@ func (detacher *cinderDiskDetacher) waitDiskDetached(instanceID, volumeID string
 
 func (detacher *cinderDiskDetacher) Detach(volumeName string, nodeName types.NodeName) error {
 	volumeID := path.Base(volumeName)
-	instances, res := detacher.cinderProvider.Instances()
-	if !res {
-		return fmt.Errorf("failed to list openstack instances")
-	}
-	instanceID, err := instances.InstanceID(nodeName)
-	if ind := strings.LastIndex(instanceID, "/"); ind >= 0 {
-		instanceID = instanceID[(ind + 1):]
-	}
-
 	if err := detacher.waitOperationFinished(volumeID); err != nil {
 		return err
 	}
-
-	attached, err := detacher.cinderProvider.DiskIsAttached(instanceID, volumeID)
+	attached, instanceID, err := detacher.cinderProvider.DiskIsAttachedByName(nodeName, volumeID)
 	if err != nil {
 		// Log error and continue with detach
 		glog.Errorf(
diff --git a/pkg/volume/cinder/attacher_test.go b/pkg/volume/cinder/attacher_test.go
diff --git a/pkg/volume/cinder/cinder.go b/pkg/volume/cinder/cinder.go

Original file line number	Diff line number	Diff line change
`@@ -537,7 +537,7 @@ func getSubnetIDForLB(compute *gophercloud.ServiceClient, node v1.Node) (string,`
`537`	`537`	`for _, intf := range interfaces {`
`538`	`538`	`for _, fixedIP := range intf.FixedIPs {`
`539`	`539`	`if fixedIP.IPAddress == ipAddress {`
`540`		`- return intf.NetID, nil`
	`540`	`+ return fixedIP.SubnetID, nil`
`541`	`541`	`}`
`542`	`542`	`}`
`543`	`543`	`}`
`@@ -551,7 +551,7 @@ func getNodeSecurityGroupIDForLB(compute gophercloud.ServiceClient, nodes []v1`
`551`	`551`
`552`	`552`	`for _, node := range nodes {`
`553`	`553`	`nodeName := types.NodeName(node.Name)`
`554`		`- srv, err := getServerByName(compute, nodeName)`
	`554`	`+ srv, err := getServerByName(compute, nodeName, true)`
`555`	`555`	`if err != nil {`
`556`	`556`	`return nodeSecurityGroupIDs.List(), err`
`557`	`557`	`}`