Skip to content

K
k3s
Unverified Commit 4afdc33c authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #56454 from mtanino/volumehandler-refactor

Automatic merge from submit-queue (batch tested with PRs 59767, 56454, 59237, 59730, 55479). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>. Block Volume: Refactor volumehandler in operationexecutor **What this PR does / why we need it**: Based on discussion with @saad-ali at #51494, we need refactor volumehandler in operationexecutor for Block Volume feature. We don't need to add volumehandler as separated object. ``` VolumeHandler does not need to be a separate object that is constructed inline like this. You can create a new operation, e.g. UnmountOperation to which you pass the spec, and it can return either a UnmountVolume or UnmapVolume. ``` **Which issue(s) this PR fixes** : no related issue. **Special notes for your reviewer**: @saad-ali @msau42 **Release note**: ```release-note NONE ```
parents 4efbb71b bc86537f
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 90 additions and 49 deletions
pkg/kubelet/volumemanager/reconciler/reconciler.go
View file @ 4afdc33c
......@@ -166,12 +166,10 @@ func (rc *reconciler) reconcile() {
// Ensure volumes that should be unmounted are unmounted.
for _, mountedVolume := range rc.actualStateOfWorld.GetMountedVolumes() {
if !rc.desiredStateOfWorld.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName) {
volumeHandler, err := operationexecutor.NewVolumeHandler(mountedVolume.VolumeSpec, rc.operationExecutor)
if err != nil {
glog.Errorf(mountedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.NewVolumeHandler for UnmountVolume failed"), err).Error())
continue
}
err = volumeHandler.UnmountVolumeHandler(mountedVolume.MountedVolume, rc.actualStateOfWorld)
// Volume is mounted, unmount it
glog.V(12).Infof(mountedVolume.GenerateMsgDetailed("Starting operationExecutor.UnmountVolume", ""))
err := rc.operationExecutor.UnmountVolume(
mountedVolume.MountedVolume, rc.actualStateOfWorld)
if err != nil &&
!nestedpendingoperations.IsAlreadyExists(err) &&
!exponentialbackoff.IsExponentialBackoff(err) {
......@@ -236,12 +234,12 @@ func (rc *reconciler) reconcile() {
if isRemount {
remountingLogStr = "Volume is already mounted to pod, but remount was requested."
}
volumeHandler, err := operationexecutor.NewVolumeHandler(volumeToMount.VolumeSpec, rc.operationExecutor)
if err != nil {
glog.Errorf(volumeToMount.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.NewVolumeHandler for MountVolume failed"), err).Error())
continue
}
err = volumeHandler.MountVolumeHandler(rc.waitForAttachTimeout, volumeToMount.VolumeToMount, rc.actualStateOfWorld, isRemount, remountingLogStr)
glog.V(12).Infof(volumeToMount.GenerateMsgDetailed("Starting operationExecutor.MountVolume", remountingLogStr))
err := rc.operationExecutor.MountVolume(
rc.waitForAttachTimeout,
volumeToMount.VolumeToMount,
rc.actualStateOfWorld,
isRemount)
if err != nil &&
!nestedpendingoperations.IsAlreadyExists(err) &&
!exponentialbackoff.IsExponentialBackoff(err) {
......@@ -265,12 +263,10 @@ func (rc *reconciler) reconcile() {
if !rc.desiredStateOfWorld.VolumeExists(attachedVolume.VolumeName) &&
!rc.operationExecutor.IsOperationPending(attachedVolume.VolumeName, nestedpendingoperations.EmptyUniquePodName) {
if attachedVolume.GloballyMounted {
volumeHandler, err := operationexecutor.NewVolumeHandler(attachedVolume.VolumeSpec, rc.operationExecutor)
if err != nil {
glog.Errorf(attachedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.NewVolumeHandler for UnmountDevice failed"), err).Error())
continue
}
err = volumeHandler.UnmountDeviceHandler(attachedVolume.AttachedVolume, rc.actualStateOfWorld, rc.mounter)
// Volume is globally mounted to device, unmount it
glog.V(12).Infof(attachedVolume.GenerateMsgDetailed("Starting operationExecutor.UnmountDevice", ""))
err := rc.operationExecutor.UnmountDevice(
attachedVolume.AttachedVolume, rc.actualStateOfWorld, rc.mounter)
if err != nil &&
!nestedpendingoperations.IsAlreadyExists(err) &&
!exponentialbackoff.IsExponentialBackoff(err) {
......@@ -403,14 +399,9 @@ func (rc *reconciler) cleanupMounts(volume podVolume) {
PluginName: volume.pluginName,
PodUID: types.UID(volume.podName),
}
volumeHandler, err := operationexecutor.NewVolumeHandlerWithMode(volume.volumeMode, rc.operationExecutor)
if err != nil {
glog.Errorf(mountedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.NewVolumeHandler for UnmountVolume failed"), err).Error())
return
}
// TODO: Currently cleanupMounts only includes UnmountVolume operation. In the next PR, we will add
// to unmount both volume and device in the same routine.
err = volumeHandler.UnmountVolumeHandler(mountedVolume, rc.actualStateOfWorld)
err := rc.operationExecutor.UnmountVolume(mountedVolume, rc.actualStateOfWorld)
if err != nil {
glog.Errorf(mountedVolume.GenerateErrorDetailed(fmt.Sprintf("volumeHandler.UnmountVolumeHandler for UnmountVolume failed"), err).Error())
return
......@@ -435,15 +426,12 @@ func (rc *reconciler) reconstructVolume(volume podVolume) (*reconstructedVolume,
UID: types.UID(volume.podName),
},
}
volumeHandler, err := operationexecutor.NewVolumeHandlerWithMode(volume.volumeMode, rc.operationExecutor)
if err != nil {
return nil, err
}
mapperPlugin, err := rc.volumePluginMgr.FindMapperPluginByName(volume.pluginName)
if err != nil {
return nil, err
}
volumeSpec, err := volumeHandler.ReconstructVolumeHandler(
volumeSpec, err := rc.operationExecutor.ReconstructVolumeOperation(
volume.volumeMode,
plugin,
mapperPlugin,
pod.UID,
......@@ -466,7 +454,7 @@ func (rc *reconciler) reconstructVolume(volume podVolume) (*reconstructedVolume,
uniqueVolumeName = volumehelper.GetUniqueVolumeNameForNonAttachableVolume(volume.podName, plugin, volumeSpec)
}
// Check existence of mount point for filesystem volume or symbolic link for block volume
isExist, checkErr := volumeHandler.CheckVolumeExistence(volume.mountPath, volumeSpec.Name(), rc.mounter, uniqueVolumeName, volume.podName, pod.UID, attachablePlugin)
isExist, checkErr := rc.operationExecutor.CheckVolumeExistenceOperation(volumeSpec, volume.mountPath, volumeSpec.Name(), rc.mounter, uniqueVolumeName, volume.podName, pod.UID, attachablePlugin)
if checkErr != nil {
return nil, err
}
......
pkg/kubelet/volumemanager/reconciler/reconciler_test.go
View file @ 4afdc33c
......@@ -835,6 +835,8 @@ func Test_GenerateMapVolumeFunc_Plugin_Not_Found(t *testing.T) {
},
}
// Enable BlockVolume feature gate
utilfeature.DefaultFeatureGate.Set("BlockVolume=true")
for name, tc := range testCases {
t.Run(name, func(t *testing.T) {
volumePluginMgr := &volume.VolumePluginMgr{}
......@@ -854,14 +856,20 @@ func Test_GenerateMapVolumeFunc_Plugin_Not_Found(t *testing.T) {
},
Spec: v1.PodSpec{},
}
volumeToMount := operationexecutor.VolumeToMount{Pod: pod, VolumeSpec: &volume.Spec{}}
err := oex.MapVolume(waitForAttachTimeout, volumeToMount, asw)
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
volumeToMount := operationexecutor.VolumeToMount{
Pod: pod,
VolumeSpec: tmpSpec}
err := oex.MountVolume(waitForAttachTimeout, volumeToMount, asw, false)
// Assert
if assert.Error(t, err) {
assert.Contains(t, err.Error(), tc.expectedErrMsg)
}
})
}
// Rollback feature gate to false.
utilfeature.DefaultFeatureGate.Set("BlockVolume=false")
}
func Test_GenerateUnmapVolumeFunc_Plugin_Not_Found(t *testing.T) {
......@@ -882,6 +890,8 @@ func Test_GenerateUnmapVolumeFunc_Plugin_Not_Found(t *testing.T) {
},
}
// Enable BlockVolume feature gate
utilfeature.DefaultFeatureGate.Set("BlockVolume=true")
for name, tc := range testCases {
t.Run(name, func(t *testing.T) {
volumePluginMgr := &volume.VolumePluginMgr{}
......@@ -893,14 +903,20 @@ func Test_GenerateUnmapVolumeFunc_Plugin_Not_Found(t *testing.T) {
nil, /* fakeRecorder */
false, /* checkNodeCapabilitiesBeforeMount */
nil))
volumeToUnmount := operationexecutor.MountedVolume{PluginName: "fake-file-plugin"}
err := oex.UnmapVolume(volumeToUnmount, asw)
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
volumeToUnmount := operationexecutor.MountedVolume{
PluginName: "fake-file-plugin",
VolumeSpec: tmpSpec}
err := oex.UnmountVolume(volumeToUnmount, asw)
// Assert
if assert.Error(t, err) {
assert.Contains(t, err.Error(), tc.expectedErrMsg)
}
})
}
// Rollback feature gate to false.
utilfeature.DefaultFeatureGate.Set("BlockVolume=false")
}
func Test_GenerateUnmapDeviceFunc_Plugin_Not_Found(t *testing.T) {
......@@ -912,15 +928,17 @@ func Test_GenerateUnmapDeviceFunc_Plugin_Not_Found(t *testing.T) {
"volumePlugin is nil": {
volumePlugins: []volume.VolumePlugin{},
expectErr: true,
expectedErrMsg: "UnmapDevice.FindMapperPluginBySpec failed",
expectedErrMsg: "UnmapDevice.FindMapperPluginByName failed",
},
"blockVolumePlugin is nil": {
volumePlugins: volumetesting.NewFakeFileVolumePlugin(),
expectErr: true,
expectedErrMsg: "UnmapDevice.FindMapperPluginBySpec failed to find BlockVolumeMapper plugin. Volume plugin is nil.",
expectedErrMsg: "UnmapDevice.FindMapperPluginByName failed to find BlockVolumeMapper plugin. Volume plugin is nil.",
},
}
// Enable BlockVolume feature gate
utilfeature.DefaultFeatureGate.Set("BlockVolume=true")
for name, tc := range testCases {
t.Run(name, func(t *testing.T) {
volumePluginMgr := &volume.VolumePluginMgr{}
......@@ -933,15 +951,18 @@ func Test_GenerateUnmapDeviceFunc_Plugin_Not_Found(t *testing.T) {
false, /* checkNodeCapabilitiesBeforeMount */
nil))
var mounter mount.Interface
plugins := volumetesting.NewFakeFileVolumePlugin()
deviceToDetach := operationexecutor.AttachedVolume{VolumeSpec: &volume.Spec{}, PluginName: plugins[0].GetPluginName()}
err := oex.UnmapDevice(deviceToDetach, asw, mounter)
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
deviceToDetach := operationexecutor.AttachedVolume{VolumeSpec: tmpSpec, PluginName: "fake-file-plugin"}
err := oex.UnmountDevice(deviceToDetach, asw, mounter)
// Assert
if assert.Error(t, err) {
assert.Contains(t, err.Error(), tc.expectedErrMsg)
}
})
}
// Rollback feature gate to false.
utilfeature.DefaultFeatureGate.Set("BlockVolume=false")
}
func waitForMount(
......
pkg/volume/util/operationexecutor/operation_executor_test.go
View file @ 4afdc33c
......@@ -231,12 +231,14 @@ func TestOperationExecutor_VerifyControllerAttachedVolumeConcurrently(t *testing
}
}
func TestOperationExecutor_MapVolume_ConcurrentMapForNonAttachablePlugins(t *testing.T) {
func TestOperationExecutor_MountVolume_ConcurrentMountForNonAttachablePlugins_VolumeMode_Block(t *testing.T) {
// Arrange
ch, quit, oe := setup()
volumesToMount := make([]VolumeToMount, numVolumesToMap)
secretName := "secret-volume"
volumeName := v1.UniqueVolumeName(secretName)
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
// Act
for i := range volumesToMount {
......@@ -247,8 +249,9 @@ func TestOperationExecutor_MapVolume_ConcurrentMapForNonAttachablePlugins(t *tes
VolumeName: volumeName,
PluginIsAttachable: false, // this field determines whether the plugin is attachable
ReportedInUse: true,
VolumeSpec: tmpSpec,
}
oe.MapVolume(0 /* waitForAttachTimeOut */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
oe.MountVolume(0 /* waitForAttachTimeOut */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */, false)
}
// Assert
......@@ -257,12 +260,14 @@ func TestOperationExecutor_MapVolume_ConcurrentMapForNonAttachablePlugins(t *tes
}
}
func TestOperationExecutor_MapVolume_ConcurrentMapForAttachablePlugins(t *testing.T) {
func TestOperationExecutor_MountVolume_ConcurrentMountForAttachablePlugins_VolumeMode_Block(t *testing.T) {
// Arrange
ch, quit, oe := setup()
volumesToMount := make([]VolumeToMount, numVolumesToAttach)
pdName := "pd-volume"
volumeName := v1.UniqueVolumeName(pdName)
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
// Act
for i := range volumesToMount {
......@@ -273,8 +278,9 @@ func TestOperationExecutor_MapVolume_ConcurrentMapForAttachablePlugins(t *testin
VolumeName: volumeName,
PluginIsAttachable: true, // this field determines whether the plugin is attachable
ReportedInUse: true,
VolumeSpec: tmpSpec,
}
oe.MapVolume(0 /* waitForAttachTimeout */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
oe.MountVolume(0 /* waitForAttachTimeout */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */, false)
}
// Assert
......@@ -283,12 +289,14 @@ func TestOperationExecutor_MapVolume_ConcurrentMapForAttachablePlugins(t *testin
}
}
func TestOperationExecutor_UnmapVolume_ConcurrentUnmapForAllPlugins(t *testing.T) {
func TestOperationExecutor_UnmountVolume_ConcurrentUnmountForAllPlugins_VolumeMode_Block(t *testing.T) {
// Arrange
ch, quit, oe := setup()
volumesToUnmount := make([]MountedVolume, numAttachableVolumesToUnmap+numNonAttachableVolumesToUnmap)
pdName := "pd-volume"
secretName := "secret-volume"
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
// Act
for i := 0; i < numNonAttachableVolumesToUnmap+numAttachableVolumesToUnmap; i++ {
......@@ -299,6 +307,7 @@ func TestOperationExecutor_UnmapVolume_ConcurrentUnmapForAllPlugins(t *testing.T
PodName: volumetypes.UniquePodName(podName),
VolumeName: v1.UniqueVolumeName(secretName),
PodUID: pod.UID,
VolumeSpec: tmpSpec,
}
} else {
pod := getTestPodWithGCEPD(podName, pdName)
......@@ -306,9 +315,10 @@ func TestOperationExecutor_UnmapVolume_ConcurrentUnmapForAllPlugins(t *testing.T
PodName: volumetypes.UniquePodName(podName),
VolumeName: v1.UniqueVolumeName(pdName),
PodUID: pod.UID,
VolumeSpec: tmpSpec,
}
}
oe.UnmapVolume(volumesToUnmount[i], nil /* actualStateOfWorldMounterUpdater */)
oe.UnmountVolume(volumesToUnmount[i], nil /* actualStateOfWorldMounterUpdater */)
}
// Assert
......@@ -317,19 +327,22 @@ func TestOperationExecutor_UnmapVolume_ConcurrentUnmapForAllPlugins(t *testing.T
}
}
func TestOperationExecutor_UnmapDeviceConcurrently(t *testing.T) {
func TestOperationExecutor_UnmountDeviceConcurrently_VolumeMode_Block(t *testing.T) {
// Arrange
ch, quit, oe := setup()
attachedVolumes := make([]AttachedVolume, numDevicesToUnmap)
pdName := "pd-volume"
volumeMode := v1.PersistentVolumeBlock
tmpSpec := &volume.Spec{PersistentVolume: &v1.PersistentVolume{Spec: v1.PersistentVolumeSpec{VolumeMode: &volumeMode}}}
// Act
for i := range attachedVolumes {
attachedVolumes[i] = AttachedVolume{
VolumeName: v1.UniqueVolumeName(pdName),
NodeName: "node-name",
VolumeSpec: tmpSpec,
}
oe.UnmapDevice(attachedVolumes[i], nil /* actualStateOfWorldMounterUpdater */, nil /* mount.Interface */)
oe.UnmountDevice(attachedVolumes[i], nil /* actualStateOfWorldMounterUpdater */, nil /* mount.Interface */)
}
// Assert
......
pkg/volume/util/operationexecutor/operation_generator.go
View file @ 4afdc33c
......@@ -1017,10 +1017,10 @@ func (og *operationGenerator) GenerateUnmapDeviceFunc(
blockVolumePlugin, err :=
og.volumePluginMgr.FindMapperPluginByName(deviceToDetach.PluginName)
if err != nil {
return volumetypes.GeneratedOperations{}, deviceToDetach.GenerateErrorDetailed("UnmapDevice.FindMapperPluginBySpec failed", err)
return volumetypes.GeneratedOperations{}, deviceToDetach.GenerateErrorDetailed("UnmapDevice.FindMapperPluginByName failed", err)
}
if blockVolumePlugin == nil {
return volumetypes.GeneratedOperations{}, deviceToDetach.GenerateErrorDetailed("UnmapDevice.FindMapperPluginBySpec failed to find BlockVolumeMapper plugin. Volume plugin is nil.", nil)
return volumetypes.GeneratedOperations{}, deviceToDetach.GenerateErrorDetailed("UnmapDevice.FindMapperPluginByName failed to find BlockVolumeMapper plugin. Volume plugin is nil.", nil)
}
blockVolumeUnmapper, newUnmapperErr := blockVolumePlugin.NewBlockVolumeUnmapper(
......
pkg/volume/util/volumehelper/BUILD
View file @ 4afdc33c
......@@ -10,10 +10,12 @@ go_library(
srcs = ["volumehelper.go"],
importpath = "k8s.io/kubernetes/pkg/volume/util/volumehelper",
deps = [
"//pkg/features:go_default_library",
"//pkg/util/mount:go_default_library",
"//pkg/volume:go_default_library",
"//pkg/volume/util/types:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
],
)
......
pkg/volume/util/volumehelper/volumehelper.go
View file @ 4afdc33c
......@@ -23,6 +23,8 @@ import (
"strings"
"k8s.io/api/core/v1"
utilfeature "k8s.io/apiserver/pkg/util/feature"
"k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/util/mount"
"k8s.io/kubernetes/pkg/volume"
"k8s.io/kubernetes/pkg/volume/util/types"
......@@ -157,3 +159,18 @@ func GetPersistentVolumeClaimVolumeMode(claim *v1.PersistentVolumeClaim) (v1.Per
}
return "", fmt.Errorf("cannot get volumeMode from pvc: %v", claim.Name)
}
// CheckVolumeModeFilesystem checks VolumeMode.
// If the mode is Filesystem, return true otherwise return false.
func CheckVolumeModeFilesystem(volumeSpec *volume.Spec) (bool, error) {
if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
volumeMode, err := GetVolumeMode(volumeSpec)
if err != nil {
return true, err
}
if volumeMode == v1.PersistentVolumeBlock {
return false, nil
}
}
return true, 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