Commit 562d0fea authored by Jing Xu's avatar Jing Xu

Handle failed attach operation leave uncertain volume attach state

This commit adds the unit tests for the PR. It also includes some files that are affected by the function name changes.
parent 47331cf0
......@@ -435,7 +435,7 @@ func (adc *attachDetachController) populateDesiredStateOfWorld() error {
err)
continue
}
if adc.actualStateOfWorld.VolumeNodeExists(volumeName, nodeName) {
if adc.actualStateOfWorld.IsVolumeAttachedToNode(volumeName, nodeName) {
devicePath, err := adc.getNodeVolumeDevicePath(volumeName, nodeName)
if err != nil {
klog.Errorf("Failed to find device path: %v", err)
......@@ -632,7 +632,7 @@ func (adc *attachDetachController) syncPVCByKey(key string) error {
func (adc *attachDetachController) processVolumesInUse(
nodeName types.NodeName, volumesInUse []v1.UniqueVolumeName) {
klog.V(4).Infof("processVolumesInUse for node %q", nodeName)
for _, attachedVolume := range adc.actualStateOfWorld.GetAttachedVolumesForNode(nodeName) {
for _, attachedVolume := range adc.actualStateOfWorld.GetAllVolumesForNode(nodeName) {
mounted := false
for _, volumeInUse := range volumesInUse {
if attachedVolume.VolumeName == volumeInUse {
......
......@@ -110,7 +110,7 @@ func Test_AttachDetachControllerStateOfWolrdPopulators_Positive(t *testing.T) {
for _, node := range nodes {
nodeName := types.NodeName(node.Name)
for _, attachedVolume := range node.Status.VolumesAttached {
found := adc.actualStateOfWorld.VolumeNodeExists(attachedVolume.Name, nodeName)
found := adc.actualStateOfWorld.IsVolumeAttachedToNode(attachedVolume.Name, nodeName)
if !found {
t.Fatalf("Run failed with error. Node %s, volume %s not found", nodeName, attachedVolume.Name)
}
......@@ -278,7 +278,7 @@ func attachDetachRecoveryTestCase(t *testing.T, extraPods1 []*v1.Pod, extraPods2
var detachedVolumesNum int = 0
time.Sleep(time.Second * 1) // Wait for a second
for _, volumeList := range testPlugin.GetAttachedVolumes() {
for _, volumeList := range testPlugin.GetAllVolumes() {
attachedVolumesNum += len(volumeList)
}
for _, volumeList := range testPlugin.GetDetachedVolumes() {
......
......@@ -34,6 +34,7 @@ go_test(
deps = [
"//pkg/controller/volume/attachdetach/testing:go_default_library",
"//pkg/volume/testing:go_default_library",
"//pkg/volume/util:go_default_library",
"//pkg/volume/util/types:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
......
......@@ -47,8 +47,10 @@ type ActualStateOfWorld interface {
// operationexecutor to interact with it.
operationexecutor.ActualStateOfWorldAttacherUpdater
// AddVolumeNode adds the given volume and node to the underlying store
// indicating the specified volume is attached to the specified node.
// AddVolumeNode adds the given volume and node to the underlying store.
// If attached is set to true, it indicates the specified volume is already
// attached to the specified node. If attached set to false, it means that
// the volume is not confirmed to be attached to the node yet.
// A unique volume name is generated from the volumeSpec and returned on
// success.
// If volumeSpec is not an attachable volume plugin, an error is returned.
......@@ -332,7 +334,7 @@ func (asw *actualStateOfWorld) AddVolumeNode(
nodeName)
}
volumeObj.nodesAttachedTo[nodeName] = node
if isAttached {
asw.addVolumeToReportAsAttached(volumeName, nodeName)
}
......
......@@ -185,7 +185,7 @@ func (collector *attachDetachStateCollector) getTotalVolumesCount() volumeCount
stateVolumeMap.add("desired_state_of_world", pluginName)
}
}
for _, v := range collector.asw.GetAttachedVolumes() {
for _, v := range collector.asw.GetAllVolumes() {
if plugin, err := collector.volumePluginMgr.FindPluginBySpec(v.VolumeSpec); err == nil {
pluginName := pluginNameNotAvailable
if plugin != nil {
......
......@@ -143,7 +143,7 @@ func TestTotalVolumesMetricCollection(t *testing.T) {
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
asw.AddVolumeNode(volumeName, volumeSpec, nodeName, "")
asw.AddVolumeNode(volumeName, volumeSpec, nodeName, "", true)
metricCollector := newAttachDetachStateCollector(
nil,
......
......@@ -175,16 +175,15 @@ func (rc *reconciler) reconcile() {
// pods that are rescheduled to a different node are detached first.
// Ensure volumes that should be detached are detached.
for _, attachedVolume := range rc.actualStateOfWorld.GetAttachedVolumes() {
for _, attachedVolume := range rc.actualStateOfWorld.GetAllVolumes() {
if !rc.desiredStateOfWorld.VolumeExists(
attachedVolume.VolumeName, attachedVolume.NodeName) {
// Don't even try to start an operation if there is already one running
// This check must be done before we do any other checks, as otherwise the other checks
// may pass while at the same time the volume leaves the pending state, resulting in
// double detach attempts
if rc.attacherDetacher.IsOperationPending(attachedVolume.VolumeName, "") {
klog.V(10).Infof("Operation for volume %q is already running. Can't start detach for %q", attachedVolume.VolumeName, attachedVolume.NodeName)
klog.V(5).Infof("Operation for volume %q is already running. Can't start detach for %q", attachedVolume.VolumeName, attachedVolume.NodeName)
continue
}
......@@ -198,7 +197,7 @@ func (rc *reconciler) reconcile() {
timeout := elapsedTime > rc.maxWaitForUnmountDuration
// Check whether volume is still mounted. Skip detach if it is still mounted unless timeout
if attachedVolume.MountedByNode && !timeout {
klog.V(12).Infof(attachedVolume.GenerateMsgDetailed("Cannot detach volume because it is still mounted", ""))
klog.V(5).Infof(attachedVolume.GenerateMsgDetailed("Cannot detach volume because it is still mounted", ""))
continue
}
......@@ -253,7 +252,7 @@ func (rc *reconciler) reconcile() {
func (rc *reconciler) attachDesiredVolumes() {
// Ensure volumes that should be attached are attached.
for _, volumeToAttach := range rc.desiredStateOfWorld.GetVolumesToAttach() {
if rc.actualStateOfWorld.VolumeNodeExists(volumeToAttach.VolumeName, volumeToAttach.NodeName) {
if rc.actualStateOfWorld.IsVolumeAttachedToNode(volumeToAttach.VolumeName, volumeToAttach.NodeName) {
// Volume/Node exists, touch it to reset detachRequestedTime
if klog.V(5) {
klog.Infof(volumeToAttach.GenerateMsgDetailed("Volume attached--touching", ""))
......
......@@ -377,7 +377,7 @@ func Test_Run_OneVolumeAttachAndDetachMultipleNodesWithReadWriteMany(t *testing.
volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
volumeSpec.PersistentVolume.Spec.AccessModes = []v1.PersistentVolumeAccessMode{v1.ReadWriteMany}
nodeName1 := k8stypes.NodeName("node-name1")
nodeName2 := k8stypes.NodeName("node-name2")
nodeName2 := k8stypes.NodeName(volumetesting.MultiAttachNode)
dsw.AddNode(nodeName1, false /*keepTerminatedPodVolumes*/)
dsw.AddNode(nodeName2, false /*keepTerminatedPodVolumes*/)
......@@ -385,6 +385,7 @@ func Test_Run_OneVolumeAttachAndDetachMultipleNodesWithReadWriteMany(t *testing.
if podAddErr != nil {
t.Fatalf("AddPod failed. Expected: <no error> Actual: <%v>", podAddErr)
}
_, podAddErr = dsw.AddPod(types.UniquePodName(podName2), controllervolumetesting.NewPod(podName2, podName2), volumeSpec, nodeName2)
if podAddErr != nil {
t.Fatalf("AddPod failed. Expected: <no error> Actual: <%v>", podAddErr)
......@@ -532,6 +533,72 @@ func Test_Run_OneVolumeAttachAndDetachMultipleNodesWithReadWriteOnce(t *testing.
waitForTotalAttachCallCount(t, 2 /* expectedAttachCallCount */, fakePlugin)
}
// Creates a volume with accessMode ReadWriteOnce
// First create a pod which will try to attach the volume to the a node named "uncertain-node". The attach call for this node will
// fail for timeout, but the volume will be actually attached to the node after the call.
// Secondly, delete the this pod.
// Lastly, create a pod scheduled to a normal node which will trigger attach volume to the node. The attach should return successfully.
func Test_Run_OneVolumeAttachAndDetachUncertainNodesWithReadWriteOnce(t *testing.T) {
// Arrange
volumePluginMgr, _ := volumetesting.GetTestVolumePluginMgr(t)
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
asw := cache.NewActualStateOfWorld(volumePluginMgr)
fakeKubeClient := controllervolumetesting.CreateTestClient()
fakeRecorder := &record.FakeRecorder{}
fakeHandler := volumetesting.NewBlockVolumePathHandler()
ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
fakeKubeClient,
volumePluginMgr,
fakeRecorder,
false, /* checkNodeCapabilitiesBeforeMount */
fakeHandler))
nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
reconciler := NewReconciler(
reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
podName1 := "pod-uid1"
podName2 := "pod-uid2"
volumeName := v1.UniqueVolumeName("volume-name")
volumeSpec := controllervolumetesting.GetTestVolumeSpec(string(volumeName), volumeName)
volumeSpec.PersistentVolume.Spec.AccessModes = []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce}
nodeName1 := k8stypes.NodeName(volumetesting.UncertainAttachNode)
nodeName2 := k8stypes.NodeName("node-name2")
dsw.AddNode(nodeName1, false /*keepTerminatedPodVolumes*/)
dsw.AddNode(nodeName2, false /*keepTerminatedPodVolumes*/)
// Act
ch := make(chan struct{})
go reconciler.Run(ch)
defer close(ch)
// Add the pod in which the volume is attached to the uncertain node
generatedVolumeName, podAddErr := dsw.AddPod(types.UniquePodName(podName1), controllervolumetesting.NewPod(podName1, podName1), volumeSpec, nodeName1)
if podAddErr != nil {
t.Fatalf("AddPod failed. Expected: <no error> Actual: <%v>", podAddErr)
}
// Volume is added to asw. Because attach operation fails, volume should not reported as attached to the node.
waitForVolumeAddedToNode(t, generatedVolumeName, nodeName1, asw)
veriryVolumeAttachedToNode(t, generatedVolumeName, nodeName1, false, asw)
veriryVolumeReportedAsAttachedToNode(t, generatedVolumeName, nodeName1, false, asw)
// When volume is added to the node, it is set to mounted by default. Then the status will be updated by checking node status VolumeInUse.
// Without this, the delete operation will be delayed due to mounted status
asw.SetVolumeMountedByNode(generatedVolumeName, nodeName1, false /* mounted */)
dsw.DeletePod(types.UniquePodName(podName1), generatedVolumeName, nodeName1)
waitForVolumeRemovedFromNode(t, generatedVolumeName, nodeName1, asw)
// Add a second pod which tries to attach the volume to a different node.
generatedVolumeName, podAddErr = dsw.AddPod(types.UniquePodName(podName2), controllervolumetesting.NewPod(podName2, podName2), volumeSpec, nodeName2)
if podAddErr != nil {
t.Fatalf("AddPod failed. Expected: <no error> Actual: <%v>", podAddErr)
}
waitForVolumeAttachedToNode(t, generatedVolumeName, nodeName2, asw)
veriryVolumeAttachedToNode(t, generatedVolumeName, nodeName2, true, asw)
}
func Test_ReportMultiAttachError(t *testing.T) {
type nodeWithPods struct {
name k8stypes.NodeName
......@@ -905,6 +972,149 @@ func verifyNewAttacherCallCount(
}
}
func waitForVolumeAttachedToNode(
t *testing.T,
volumeName v1.UniqueVolumeName,
nodeName k8stypes.NodeName,
asw cache.ActualStateOfWorld) {
err := retryWithExponentialBackOff(
time.Duration(500*time.Millisecond),
func() (bool, error) {
if asw.IsVolumeAttachedToNode(volumeName, nodeName) {
return true, nil
}
t.Logf(
"Warning: Volume <%v> is not attached to node <%v> yet. Will retry.",
volumeName,
nodeName)
return false, nil
},
)
if err != nil && !asw.IsVolumeAttachedToNode(volumeName, nodeName) {
t.Fatalf(
"Volume <%v> is not attached to node <%v>.",
volumeName,
nodeName)
}
}
func waitForVolumeAddedToNode(
t *testing.T,
volumeName v1.UniqueVolumeName,
nodeName k8stypes.NodeName,
asw cache.ActualStateOfWorld) {
err := retryWithExponentialBackOff(
time.Duration(500*time.Millisecond),
func() (bool, error) {
volumes := asw.GetAllVolumes()
for _, volume := range volumes {
if volume.VolumeName == volumeName && volume.NodeName == nodeName {
return true, nil
}
}
t.Logf(
"Warning: Volume <%v> is not added to node <%v> yet. Will retry.",
volumeName,
nodeName)
return false, nil
},
)
if err != nil {
t.Fatalf(
"Volume <%v> is not added to node <%v>. %v",
volumeName,
nodeName, err)
}
}
func waitForVolumeRemovedFromNode(
t *testing.T,
volumeName v1.UniqueVolumeName,
nodeName k8stypes.NodeName,
asw cache.ActualStateOfWorld) {
err := retryWithExponentialBackOff(
time.Duration(500*time.Millisecond),
func() (bool, error) {
volumes := asw.GetAllVolumes()
exist := false
for _, volume := range volumes {
if volume.VolumeName == volumeName && volume.NodeName == nodeName {
exist = true
}
}
if exist {
t.Logf(
"Warning: Volume <%v> is not removed from the node <%v> yet. Will retry.",
volumeName,
nodeName)
return false, nil
}
return true, nil
},
)
if err != nil {
t.Fatalf(
"Volume <%v> is not removed from node <%v>. %v",
volumeName,
nodeName, err)
}
}
func veriryVolumeAttachedToNode(
t *testing.T,
volumeName v1.UniqueVolumeName,
nodeName k8stypes.NodeName,
isAttached bool,
asw cache.ActualStateOfWorld,
) {
result := asw.IsVolumeAttachedToNode(volumeName, nodeName)
if result == isAttached {
return
}
t.Fatalf("Check volume <%v> is attached to node <%v>, got %v, expected %v",
volumeName,
nodeName,
result,
isAttached)
}
func veriryVolumeReportedAsAttachedToNode(
t *testing.T,
volumeName v1.UniqueVolumeName,
nodeName k8stypes.NodeName,
isAttached bool,
asw cache.ActualStateOfWorld,
) {
result := false
volumes := asw.GetVolumesToReportAttached()
for _, volume := range volumes[nodeName] {
if volume.Name == volumeName {
result = true
}
}
if result == isAttached {
return
}
t.Fatalf("Check volume <%v> is reported as attached to node <%v>, got %v, expected %v",
volumeName,
nodeName,
result,
isAttached)
}
func verifyNewDetacherCallCount(
t *testing.T,
expectZeroNewDetacherCallCount bool,
......
......@@ -328,7 +328,7 @@ func (plugin *TestPlugin) GetErrorEncountered() bool {
return plugin.ErrorEncountered
}
func (plugin *TestPlugin) GetAttachedVolumes() map[string][]string {
func (plugin *TestPlugin) GetAllVolumes() map[string][]string {
plugin.pluginLock.RLock()
defer plugin.pluginLock.RUnlock()
ret := make(map[string][]string)
......
......@@ -308,6 +308,11 @@ func (asw *actualStateOfWorld) MarkVolumeAsAttached(
return asw.addVolume(volumeName, volumeSpec, devicePath)
}
func (asw *actualStateOfWorld) MarkVolumeAsUncertain(
volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, _ types.NodeName) error {
return nil
}
func (asw *actualStateOfWorld) MarkVolumeAsDetached(
volumeName v1.UniqueVolumeName, nodeName types.NodeName) {
asw.DeleteVolume(volumeName)
......
......@@ -46,9 +46,16 @@ import (
"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
)
// A hook specified in storage class to indicate it's provisioning
// is expected to fail.
const ExpectProvisionFailureKey = "expect-provision-failure"
const (
// A hook specified in storage class to indicate it's provisioning
// is expected to fail.
ExpectProvisionFailureKey = "expect-provision-failure"
// The node is marked as uncertain. The attach operation will fail and return timeout error
// but the operation is actually succeeded.
UncertainAttachNode = "uncertain-attach-node"
// The node is marked as multi-attach which means it is allowed to attach the volume to multiple nodes.
MultiAttachNode = "multi-attach-node"
)
// fakeVolumeHost is useful for testing volume plugins.
type fakeVolumeHost struct {
......@@ -273,10 +280,15 @@ var _ DeviceMountableVolumePlugin = &FakeVolumePlugin{}
var _ FSResizableVolumePlugin = &FakeVolumePlugin{}
func (plugin *FakeVolumePlugin) getFakeVolume(list *[]*FakeVolume) *FakeVolume {
volumeList := *list
if list != nil && len(volumeList) > 0 {
return volumeList[0]
}
volume := &FakeVolume{
WaitForAttachHook: plugin.WaitForAttachHook,
UnmountDeviceHook: plugin.UnmountDeviceHook,
}
volume.VolumesAttached = make(map[string]types.NodeName)
*list = append(*list, volume)
return volume
}
......@@ -329,6 +341,8 @@ func (plugin *FakeVolumePlugin) NewMounter(spec *Spec, pod *v1.Pod, opts VolumeO
plugin.Lock()
defer plugin.Unlock()
volume := plugin.getFakeVolume(&plugin.Mounters)
volume.Lock()
defer volume.Unlock()
volume.PodUID = pod.UID
volume.VolName = spec.Name()
volume.Plugin = plugin
......@@ -346,6 +360,8 @@ func (plugin *FakeVolumePlugin) NewUnmounter(volName string, podUID types.UID) (
plugin.Lock()
defer plugin.Unlock()
volume := plugin.getFakeVolume(&plugin.Unmounters)
volume.Lock()
defer volume.Unlock()
volume.PodUID = podUID
volume.VolName = volName
volume.Plugin = plugin
......@@ -364,6 +380,8 @@ func (plugin *FakeVolumePlugin) NewBlockVolumeMapper(spec *Spec, pod *v1.Pod, op
plugin.Lock()
defer plugin.Unlock()
volume := plugin.getFakeVolume(&plugin.BlockVolumeMappers)
volume.Lock()
defer volume.Unlock()
if pod != nil {
volume.PodUID = pod.UID
}
......@@ -384,6 +402,8 @@ func (plugin *FakeVolumePlugin) NewBlockVolumeUnmapper(volName string, podUID ty
plugin.Lock()
defer plugin.Unlock()
volume := plugin.getFakeVolume(&plugin.BlockVolumeUnmappers)
volume.Lock()
defer volume.Unlock()
volume.PodUID = podUID
volume.VolName = volName
volume.Plugin = plugin
......@@ -424,7 +444,16 @@ func (plugin *FakeVolumePlugin) NewDetacher() (Detacher, error) {
plugin.Lock()
defer plugin.Unlock()
plugin.NewDetacherCallCount = plugin.NewDetacherCallCount + 1
return plugin.getFakeVolume(&plugin.Detachers), nil
detacher := plugin.getFakeVolume(&plugin.Detachers)
attacherList := plugin.Attachers
if attacherList != nil && len(attacherList) > 0 {
detacherList := plugin.Detachers
if detacherList != nil && len(detacherList) > 0 {
detacherList[0].VolumesAttached = attacherList[0].VolumesAttached
}
}
return detacher, nil
}
func (plugin *FakeVolumePlugin) NewDeviceUnmounter() (DeviceUnmounter, error) {
......@@ -557,6 +586,7 @@ type FakeVolume struct {
VolName string
Plugin *FakeVolumePlugin
MetricsNil
VolumesAttached map[string]types.NodeName
// Add callbacks as needed
WaitForAttachHook func(spec *Spec, devicePath string, pod *v1.Pod, spectimeout time.Duration) (string, error)
......@@ -577,6 +607,20 @@ type FakeVolume struct {
PodDeviceMapPathCallCount int
}
func getUniqueVolumeName(spec *Spec) (string, error) {
var volumeName string
if spec.Volume != nil && spec.Volume.GCEPersistentDisk != nil {
volumeName = spec.Volume.GCEPersistentDisk.PDName
} else if spec.PersistentVolume != nil &&
spec.PersistentVolume.Spec.GCEPersistentDisk != nil {
volumeName = spec.PersistentVolume.Spec.GCEPersistentDisk.PDName
}
if volumeName == "" {
volumeName = spec.Name()
}
return volumeName, nil
}
func (_ *FakeVolume) GetAttributes() Attributes {
return Attributes{
ReadOnly: false,
......@@ -722,6 +766,25 @@ func (fv *FakeVolume) Attach(spec *Spec, nodeName types.NodeName) (string, error
fv.Lock()
defer fv.Unlock()
fv.AttachCallCount++
volumeName, err := getUniqueVolumeName(spec)
if err != nil {
return "", err
}
volumeNode, exist := fv.VolumesAttached[volumeName]
if exist {
if nodeName == UncertainAttachNode {
return "", fmt.Errorf("Timed out to attach volume %q to node %q", volumeName, nodeName)
}
if volumeNode == nodeName || volumeNode == MultiAttachNode || nodeName == MultiAttachNode {
return "/dev/vdb-test", nil
}
return "", fmt.Errorf("volume %q trying to attach to node %q is already attached to node %q", volumeName, nodeName, volumeNode)
}
fv.VolumesAttached[volumeName] = nodeName
if nodeName == UncertainAttachNode {
return "", fmt.Errorf("Timed out to attach volume %q to node %q", volumeName, nodeName)
}
return "/dev/vdb-test", nil
}
......@@ -771,6 +834,10 @@ func (fv *FakeVolume) Detach(volumeName string, nodeName types.NodeName) error {
fv.Lock()
defer fv.Unlock()
fv.DetachCallCount++
if _, exist := fv.VolumesAttached[volumeName]; !exist {
return fmt.Errorf("Trying to detach volume %q that is not attached to the node %q", volumeName, nodeName)
}
delete(fv.VolumesAttached, volumeName)
return nil
}
......@@ -937,7 +1004,7 @@ func VerifyAttachCallCount(
fakeVolumePlugin *FakeVolumePlugin) error {
for _, attacher := range fakeVolumePlugin.GetAttachers() {
actualCallCount := attacher.GetAttachCallCount()
if actualCallCount == expectedAttachCallCount {
if actualCallCount >= expectedAttachCallCount {
return nil
}
}
......@@ -970,7 +1037,7 @@ func VerifyWaitForAttachCallCount(
fakeVolumePlugin *FakeVolumePlugin) error {
for _, attacher := range fakeVolumePlugin.GetAttachers() {
actualCallCount := attacher.GetWaitForAttachCallCount()
if actualCallCount == expectedWaitForAttachCallCount {
if actualCallCount >= expectedWaitForAttachCallCount {
return nil
}
}
......@@ -1003,7 +1070,7 @@ func VerifyMountDeviceCallCount(
fakeVolumePlugin *FakeVolumePlugin) error {
for _, attacher := range fakeVolumePlugin.GetAttachers() {
actualCallCount := attacher.GetMountDeviceCallCount()
if actualCallCount == expectedMountDeviceCallCount {
if actualCallCount >= expectedMountDeviceCallCount {
return nil
}
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment