github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-25 04:33:26 +00:00
Code Issues Projects Releases Wiki Activity

Make volume attachment watch timeout configurable for csiAttacher

Browse Source
This commit is contained in:
Chris Henzie 2020-12-28 12:22:10 -08:00
parent b4e3f272b9
commit 195a5af23b
5 changed files with 63 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
pkg/volume/csi/csi_attacher.go
View File

@ -46,8 +46,9 @@ const (
) )
type csiAttacher struct { type csiAttacher struct {
plugin *csiPlugin plugin *csiPlugin
k8s kubernetes.Interface k8s kubernetes.Interface
watchTimeout time.Duration
csiClient csiClient csiClient csiClient
} }
@ -120,7 +121,7 @@ func (c *csiAttacher) Attach(spec *volume.Spec, nodeName types.NodeName) (string
klog.V(4).Info(log("attachment [%v] for volume [%v] created successfully", attachID, pvSrc.VolumeHandle)) klog.V(4).Info(log("attachment [%v] for volume [%v] created successfully", attachID, pvSrc.VolumeHandle))
} }
if _, err := c.waitForVolumeAttachment(pvSrc.VolumeHandle, attachID, csiTimeout); err != nil { if _, err := c.waitForVolumeAttachment(pvSrc.VolumeHandle, attachID, c.watchTimeout); err != nil {
return "", err return "", err
} }
@ -256,7 +257,7 @@ func (c *csiAttacher) MountDevice(spec *volume.Spec, devicePath string, deviceMo
} }
csi := c.csiClient csi := c.csiClient
ctx, cancel := context.WithTimeout(context.Background(), csiTimeout) ctx, cancel := context.WithTimeout(context.Background(), c.watchTimeout)
defer cancel() defer cancel()
// Check whether "STAGE_UNSTAGE_VOLUME" is set // Check whether "STAGE_UNSTAGE_VOLUME" is set
stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx) stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)
@ -397,7 +398,7 @@ func (c *csiAttacher) Detach(volumeName string, nodeName types.NodeName) error {
} }
klog.V(4).Info(log("detacher deleted ok VolumeAttachment.ID=%s", attachID)) klog.V(4).Info(log("detacher deleted ok VolumeAttachment.ID=%s", attachID))
err := c.waitForVolumeDetachment(volID, attachID, csiTimeout) err := c.waitForVolumeDetachment(volID, attachID, c.watchTimeout)
return err return err
} }
@ -514,7 +515,7 @@ func (c *csiAttacher) UnmountDevice(deviceMountPath string) error {
} }
csi := c.csiClient csi := c.csiClient
ctx, cancel := context.WithTimeout(context.Background(), csiTimeout) ctx, cancel := context.WithTimeout(context.Background(), c.watchTimeout)
defer cancel() defer cancel()
// Check whether "STAGE_UNSTAGE_VOLUME" is set // Check whether "STAGE_UNSTAGE_VOLUME" is set
stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx) stageUnstageSet, err := csi.NodeSupportsStageUnstage(ctx)

54
pkg/volume/csi/csi_attacher_test.go
View File

@ -52,6 +52,8 @@ import (
volumetypes "k8s.io/kubernetes/pkg/volume/util/types" volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
) )
const testWatchTimeout = 10 * time.Second
var ( var (
bFalse = false bFalse = false
bTrue = true bTrue = true
@ -209,7 +211,7 @@ func TestAttacherAttach(t *testing.T) {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
// FIXME: We need to ensure this goroutine exits in the test. // FIXME: We need to ensure this goroutine exits in the test.
go func(spec *volume.Spec, nodename string, fail bool) { go func(spec *volume.Spec, nodename string, fail bool) {
@ -293,7 +295,7 @@ func TestAttacherAttachWithInline(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
// FIXME: We need to ensure this goroutine exits in the test. // FIXME: We need to ensure this goroutine exits in the test.
go func(spec *volume.Spec, nodename string, fail bool) { go func(spec *volume.Spec, nodename string, fail bool) {
@ -378,7 +380,7 @@ func TestAttacherWithCSIDriver(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
spec := volume.NewSpecFromPersistentVolume(makeTestPV("test-pv", 10, test.driver, "test-vol"), false) spec := volume.NewSpecFromPersistentVolume(makeTestPV("test-pv", 10, test.driver, "test-vol"), false)
pluginCanAttach, err := plug.CanAttach(spec) pluginCanAttach, err := plug.CanAttach(spec)
@ -473,7 +475,7 @@ func TestAttacherWaitForVolumeAttachmentWithCSIDriver(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
spec := volume.NewSpecFromPersistentVolume(makeTestPV("test-pv", 10, test.driver, "test-vol"), false) spec := volume.NewSpecFromPersistentVolume(makeTestPV("test-pv", 10, test.driver, "test-vol"), false)
pluginCanAttach, err := plug.CanAttach(spec) pluginCanAttach, err := plug.CanAttach(spec)
@ -547,7 +549,7 @@ func TestAttacherWaitForAttach(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
if test.makeAttachment != nil { if test.makeAttachment != nil {
attachment := test.makeAttachment() attachment := test.makeAttachment()
@ -629,7 +631,7 @@ func TestAttacherWaitForAttachWithInline(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
if test.makeAttachment != nil { if test.makeAttachment != nil {
attachment := test.makeAttachment() attachment := test.makeAttachment()
@ -716,7 +718,7 @@ func TestAttacherWaitForVolumeAttachment(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
t.Logf("running test: %v", tc.name) t.Logf("running test: %v", tc.name)
pvName := fmt.Sprintf("test-pv-%d", i) pvName := fmt.Sprintf("test-pv-%d", i)
volID := fmt.Sprintf("test-vol-%d", i) volID := fmt.Sprintf("test-vol-%d", i)
@ -812,7 +814,7 @@ func TestAttacherVolumesAreAttached(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
nodeName := "fakeNode" nodeName := "fakeNode"
var specs []*volume.Spec var specs []*volume.Spec
@ -883,7 +885,7 @@ func TestAttacherVolumesAreAttachedWithInline(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
nodeName := "fakeNode" nodeName := "fakeNode"
var specs []*volume.Spec var specs []*volume.Spec
@ -976,7 +978,7 @@ func TestAttacherDetach(t *testing.T) {
if err0 != nil { if err0 != nil {
t.Fatalf("failed to create new attacher: %v", err0) t.Fatalf("failed to create new attacher: %v", err0)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
pv := makeTestPV("test-pv", 10, testDriver, tc.volID) pv := makeTestPV("test-pv", 10, testDriver, tc.volID)
spec := volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly) spec := volume.NewSpecFromPersistentVolume(pv, pv.Spec.PersistentVolumeSource.CSI.ReadOnly)
@ -1028,7 +1030,7 @@ func TestAttacherGetDeviceMountPath(t *testing.T) {
if err0 != nil { if err0 != nil {
t.Fatalf("failed to create new attacher: %v", err0) t.Fatalf("failed to create new attacher: %v", err0)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
pluginDir := csiAttacher.plugin.host.GetPluginDir(plug.GetPluginName()) pluginDir := csiAttacher.plugin.host.GetPluginDir(plug.GetPluginName())
@ -1193,7 +1195,7 @@ func TestAttacherMountDevice(t *testing.T) {
if err0 != nil { if err0 != nil {
t.Fatalf("failed to create new attacher: %v", err0) t.Fatalf("failed to create new attacher: %v", err0)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet) csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet)
if tc.deviceMountPath != "" { if tc.deviceMountPath != "" {
@ -1356,7 +1358,7 @@ func TestAttacherMountDeviceWithInline(t *testing.T) {
if err0 != nil { if err0 != nil {
t.Fatalf("failed to create new attacher: %v", err0) t.Fatalf("failed to create new attacher: %v", err0)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet) csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet)
if tc.deviceMountPath != "" { if tc.deviceMountPath != "" {
@ -1484,7 +1486,7 @@ func TestAttacherUnmountDevice(t *testing.T) {
if err0 != nil { if err0 != nil {
t.Fatalf("failed to create new attacher: %v", err0) t.Fatalf("failed to create new attacher: %v", err0)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet) csiAttacher.csiClient = setupClient(t, tc.stageUnstageSet)
if tc.deviceMountPath != "" { if tc.deviceMountPath != "" {
@ -1630,3 +1632,27 @@ func newTestWatchPlugin(t *testing.T, fakeClient *fakeclient.Clientset, setupInf
return csiPlug, fakeWatcher, tmpDir, fakeClient return csiPlug, fakeWatcher, tmpDir, fakeClient
} }
func getCsiAttacherFromVolumeAttacher(attacher volume.Attacher) *csiAttacher {
csiAttacher := attacher.(*csiAttacher)
csiAttacher.watchTimeout = testWatchTimeout
return csiAttacher
}
func getCsiAttacherFromVolumeDetacher(detacher volume.Detacher) *csiAttacher {
csiAttacher := detacher.(*csiAttacher)
csiAttacher.watchTimeout = testWatchTimeout
return csiAttacher
}
func getCsiAttacherFromDeviceMounter(deviceMounter volume.DeviceMounter) *csiAttacher {
csiAttacher := deviceMounter.(*csiAttacher)
csiAttacher.watchTimeout = testWatchTimeout
return csiAttacher
}
func getCsiAttacherFromDeviceUnmounter(deviceUnmounter volume.DeviceUnmounter) *csiAttacher {
csiAttacher := deviceUnmounter.(*csiAttacher)
csiAttacher.watchTimeout = testWatchTimeout
return csiAttacher
}

5
pkg/volume/csi/csi_plugin.go
View File

@ -944,8 +944,9 @@ func (p *csiPlugin) newAttacherDetacher() (*csiAttacher, error) {
} }
return &csiAttacher{ return &csiAttacher{
plugin: p, plugin: p,
k8s: k8s, k8s: k8s,
watchTimeout: csiTimeout,
}, nil }, nil
} }

11
pkg/volume/csi/csi_plugin_test.go
View File

@ -22,6 +22,7 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"testing" "testing"
"time"
api "k8s.io/api/core/v1" api "k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
@ -881,13 +882,16 @@ func TestPluginNewAttacher(t *testing.T) {
t.Fatalf("failed to create new attacher: %v", err) t.Fatalf("failed to create new attacher: %v", err)
} }
csiAttacher := attacher.(*csiAttacher) csiAttacher := getCsiAttacherFromVolumeAttacher(attacher)
if csiAttacher.plugin == nil { if csiAttacher.plugin == nil {
t.Error("plugin not set for attacher") t.Error("plugin not set for attacher")
} }
if csiAttacher.k8s == nil { if csiAttacher.k8s == nil {
t.Error("Kubernetes client not set for attacher") t.Error("Kubernetes client not set for attacher")
} }
if csiAttacher.watchTimeout == time.Duration(0) {
t.Error("watch timeout not set for attacher")
}
} }
func TestPluginNewDetacher(t *testing.T) { func TestPluginNewDetacher(t *testing.T) {
@ -901,13 +905,16 @@ func TestPluginNewDetacher(t *testing.T) {
t.Fatalf("failed to create new detacher: %v", err) t.Fatalf("failed to create new detacher: %v", err)
} }
csiDetacher := detacher.(*csiAttacher) csiDetacher := getCsiAttacherFromVolumeDetacher(detacher)
if csiDetacher.plugin == nil { if csiDetacher.plugin == nil {
t.Error("plugin not set for detacher") t.Error("plugin not set for detacher")
} }
if csiDetacher.k8s == nil { if csiDetacher.k8s == nil {
t.Error("Kubernetes client not set for detacher") t.Error("Kubernetes client not set for detacher")
} }
if csiDetacher.watchTimeout == time.Duration(0) {
t.Error("watch timeout not set for detacher")
}
} }
func TestPluginCanAttach(t *testing.T) { func TestPluginCanAttach(t *testing.T) {

8
pkg/volume/csi/csi_test.go
View File

@ -387,7 +387,7 @@ func TestCSI_VolumeAll(t *testing.T) {
} }
if devMounter != nil { if devMounter != nil {
csiDevMounter := devMounter.(*csiAttacher) csiDevMounter := getCsiAttacherFromDeviceMounter(devMounter)
csiDevMounter.csiClient = csiClient csiDevMounter.csiClient = csiClient
devMountPath, err := csiDevMounter.GetDeviceMountPath(volSpec) devMountPath, err := csiDevMounter.GetDeviceMountPath(volSpec)
if err != nil { if err != nil {
@ -550,8 +550,8 @@ func TestCSI_VolumeAll(t *testing.T) {
} }
if devMounter != nil && devUnmounter != nil { if devMounter != nil && devUnmounter != nil {
csiDevMounter := devMounter.(*csiAttacher) csiDevMounter := getCsiAttacherFromDeviceMounter(devMounter)
csiDevUnmounter := devUnmounter.(*csiAttacher) csiDevUnmounter := getCsiAttacherFromDeviceUnmounter(devUnmounter)
csiDevUnmounter.csiClient = csiClient csiDevUnmounter.csiClient = csiClient
devMountPath, err := csiDevMounter.GetDeviceMountPath(volSpec) devMountPath, err := csiDevMounter.GetDeviceMountPath(volSpec)
@ -592,7 +592,7 @@ func TestCSI_VolumeAll(t *testing.T) {
if err != nil { if err != nil {
t.Fatal("csiTest.VolumeAll volumePlugin.GetVolumeName failed:", err) t.Fatal("csiTest.VolumeAll volumePlugin.GetVolumeName failed:", err)
} }
csiDetacher := volDetacher.(*csiAttacher) csiDetacher := getCsiAttacherFromVolumeDetacher(volDetacher)
csiDetacher.csiClient = csiClient csiDetacher.csiClient = csiClient
if err := csiDetacher.Detach(volName, host.GetNodeName()); err != nil { if err := csiDetacher.Detach(volName, host.GetNodeName()); err != nil {
t.Fatal("csiTest.VolumeAll detacher.Detach failed:", err) t.Fatal("csiTest.VolumeAll detacher.Detach failed:", err)