mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-09-23 10:58:42 +00:00
Reorganize volume controllers and manager
This commit is contained in:
saadali
321
pkg/kubelet/volumemanager/reconciler/reconciler.go
Normal file
321
pkg/kubelet/volumemanager/reconciler/reconciler.go
Normal file
@@ -0,0 +1,321 @@
|
||||
/*
|
||||
Copyright 2016 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
// Package reconciler implements interfaces that attempt to reconcile the
|
||||
// desired state of the with the actual state of the world by triggering
|
||||
// relevant actions (attach, detach, mount, unmount).
|
||||
package reconciler
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/golang/glog"
|
||||
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||
"k8s.io/kubernetes/pkg/kubelet/volumemanager/cache"
|
||||
"k8s.io/kubernetes/pkg/util/goroutinemap"
|
||||
"k8s.io/kubernetes/pkg/util/wait"
|
||||
"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
|
||||
)
|
||||
|
||||
// Reconciler runs a periodic loop to reconcile the desired state of the world
|
||||
// with the actual state of the world by triggering attach, detach, mount, and
|
||||
// unmount operations.
|
||||
// Note: This is distinct from the Reconciler implemented by the attach/detach
|
||||
// controller. This reconciles state for the kubelet volume manager. That
|
||||
// reconciles state for the attach/detach controller.
|
||||
type Reconciler interface {
|
||||
// Starts running the reconciliation loop which executes periodically, checks
|
||||
// if volumes that should be mounted are mounted and volumes that should
|
||||
// be unmounted are unmounted. If not, it will trigger mount/unmount
|
||||
// operations to rectify.
|
||||
// If attach/detach management is enabled, the manager will also check if
|
||||
// volumes that should be attached are attached and volumes that should
|
||||
// be detached are detached and trigger attach/detach operations as needed.
|
||||
Run(stopCh <-chan struct{})
|
||||
}
|
||||
|
||||
// NewReconciler returns a new instance of Reconciler.
|
||||
//
|
||||
// controllerAttachDetachEnabled - if true, indicates that the attach/detach
|
||||
// controller is responsible for managing the attach/detach operations for
|
||||
// this node, and therefore the volume manager should not
|
||||
// loopSleepDuration - the amount of time the reconciler loop sleeps between
|
||||
// successive executions
|
||||
// waitForAttachTimeout - the amount of time the Mount function will wait for
|
||||
// the volume to be attached
|
||||
// hostName - the hostname for this node, used by Attach and Detach methods
|
||||
// desiredStateOfWorld - cache containing the desired state of the world
|
||||
// actualStateOfWorld - cache containing the actual state of the world
|
||||
// operationExecutor - used to trigger attach/detach/mount/unmount operations
|
||||
// safely (prevents more than one operation from being triggered on the same
|
||||
// volume)
|
||||
func NewReconciler(
|
||||
kubeClient internalclientset.Interface,
|
||||
controllerAttachDetachEnabled bool,
|
||||
loopSleepDuration time.Duration,
|
||||
waitForAttachTimeout time.Duration,
|
||||
hostName string,
|
||||
desiredStateOfWorld cache.DesiredStateOfWorld,
|
||||
actualStateOfWorld cache.ActualStateOfWorld,
|
||||
operationExecutor operationexecutor.OperationExecutor) Reconciler {
|
||||
return &reconciler{
|
||||
kubeClient: kubeClient,
|
||||
controllerAttachDetachEnabled: controllerAttachDetachEnabled,
|
||||
loopSleepDuration: loopSleepDuration,
|
||||
waitForAttachTimeout: waitForAttachTimeout,
|
||||
hostName: hostName,
|
||||
desiredStateOfWorld: desiredStateOfWorld,
|
||||
actualStateOfWorld: actualStateOfWorld,
|
||||
operationExecutor: operationExecutor,
|
||||
}
|
||||
}
|
||||
|
||||
type reconciler struct {
|
||||
kubeClient internalclientset.Interface
|
||||
controllerAttachDetachEnabled bool
|
||||
loopSleepDuration time.Duration
|
||||
waitForAttachTimeout time.Duration
|
||||
hostName string
|
||||
desiredStateOfWorld cache.DesiredStateOfWorld
|
||||
actualStateOfWorld cache.ActualStateOfWorld
|
||||
operationExecutor operationexecutor.OperationExecutor
|
||||
}
|
||||
|
||||
func (rc *reconciler) Run(stopCh <-chan struct{}) {
|
||||
wait.Until(rc.reconciliationLoopFunc(), rc.loopSleepDuration, stopCh)
|
||||
}
|
||||
|
||||
func (rc *reconciler) reconciliationLoopFunc() func() {
|
||||
return func() {
|
||||
// Unmounts are triggered before mounts so that a volume that was
|
||||
// referenced by a pod that was deleted and is now referenced by another
|
||||
// pod is unmounted from the first pod before being mounted to the new
|
||||
// pod.
|
||||
|
||||
// Ensure volumes that should be unmounted are unmounted.
|
||||
for _, mountedVolume := range rc.actualStateOfWorld.GetMountedVolumes() {
|
||||
if !rc.desiredStateOfWorld.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName) {
|
||||
// Volume is mounted, unmount it
|
||||
glog.V(12).Infof("Attempting to start UnmountVolume for volume %q (spec.Name: %q) from pod %q (UID: %q).",
|
||||
mountedVolume.VolumeName,
|
||||
mountedVolume.OuterVolumeSpecName,
|
||||
mountedVolume.PodName,
|
||||
mountedVolume.PodUID)
|
||||
err := rc.operationExecutor.UnmountVolume(
|
||||
mountedVolume.MountedVolume, rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.UnmountVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
mountedVolume.VolumeName,
|
||||
mountedVolume.OuterVolumeSpecName,
|
||||
mountedVolume.PodName,
|
||||
mountedVolume.PodUID,
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("UnmountVolume operation started for volume %q (spec.Name: %q) from pod %q (UID: %q).",
|
||||
mountedVolume.VolumeName,
|
||||
mountedVolume.OuterVolumeSpecName,
|
||||
mountedVolume.PodName,
|
||||
mountedVolume.PodUID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure volumes that should be attached/mounted are attached/mounted.
|
||||
for _, volumeToMount := range rc.desiredStateOfWorld.GetVolumesToMount() {
|
||||
volMounted, devicePath, err := rc.actualStateOfWorld.PodExistsInVolume(volumeToMount.PodName, volumeToMount.VolumeName)
|
||||
volumeToMount.DevicePath = devicePath
|
||||
if cache.IsVolumeNotAttachedError(err) {
|
||||
if rc.controllerAttachDetachEnabled || !volumeToMount.PluginIsAttachable {
|
||||
// Volume is not attached (or doesn't implement attacher), kubelet attach is disabled, wait
|
||||
// for controller to finish attaching volume.
|
||||
glog.V(12).Infof("Attempting to start VerifyControllerAttachedVolume for volume %q (spec.Name: %q) pod %q (UID: %q)",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID)
|
||||
err := rc.operationExecutor.VerifyControllerAttachedVolume(
|
||||
volumeToMount.VolumeToMount,
|
||||
rc.hostName,
|
||||
rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.VerifyControllerAttachedVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID,
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("VerifyControllerAttachedVolume operation started for volume %q (spec.Name: %q) pod %q (UID: %q)",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID)
|
||||
}
|
||||
} else {
|
||||
// Volume is not attached to node, kubelet attach is enabled, volume implements an attacher,
|
||||
// so attach it
|
||||
volumeToAttach := operationexecutor.VolumeToAttach{
|
||||
VolumeName: volumeToMount.VolumeName,
|
||||
VolumeSpec: volumeToMount.VolumeSpec,
|
||||
NodeName: rc.hostName,
|
||||
}
|
||||
glog.V(12).Infof("Attempting to start AttachVolume for volume %q (spec.Name: %q) pod %q (UID: %q)",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID)
|
||||
err := rc.operationExecutor.AttachVolume(volumeToAttach, rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.AttachVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID,
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("AttachVolume operation started for volume %q (spec.Name: %q) pod %q (UID: %q)",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID)
|
||||
}
|
||||
}
|
||||
} else if !volMounted || cache.IsRemountRequiredError(err) {
|
||||
// Volume is not mounted, or is already mounted, but requires remounting
|
||||
remountingLogStr := ""
|
||||
if cache.IsRemountRequiredError(err) {
|
||||
remountingLogStr = "Volume is already mounted to pod, but remount was requested."
|
||||
}
|
||||
glog.V(12).Infof("Attempting to start MountVolume for volume %q (spec.Name: %q) to pod %q (UID: %q). %s",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID,
|
||||
remountingLogStr)
|
||||
err := rc.operationExecutor.MountVolume(
|
||||
rc.waitForAttachTimeout,
|
||||
volumeToMount.VolumeToMount,
|
||||
rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.MountVolume failed for volume %q (spec.Name: %q) pod %q (UID: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID,
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("MountVolume operation started for volume %q (spec.Name: %q) to pod %q (UID: %q). %s",
|
||||
volumeToMount.VolumeName,
|
||||
volumeToMount.VolumeSpec.Name(),
|
||||
volumeToMount.PodName,
|
||||
volumeToMount.Pod.UID,
|
||||
remountingLogStr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure devices that should be detached/unmounted are detached/unmounted.
|
||||
for _, attachedVolume := range rc.actualStateOfWorld.GetUnmountedVolumes() {
|
||||
if !rc.desiredStateOfWorld.VolumeExists(attachedVolume.VolumeName) {
|
||||
if attachedVolume.GloballyMounted {
|
||||
// Volume is globally mounted to device, unmount it
|
||||
glog.V(12).Infof("Attempting to start UnmountDevice for volume %q (spec.Name: %q)",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name())
|
||||
err := rc.operationExecutor.UnmountDevice(
|
||||
attachedVolume.AttachedVolume, rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists and goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.UnmountDevice failed for volume %q (spec.Name: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name(),
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("UnmountDevice operation started for volume %q (spec.Name: %q)",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name())
|
||||
}
|
||||
} else {
|
||||
// Volume is attached to node, detach it
|
||||
if rc.controllerAttachDetachEnabled || !attachedVolume.PluginIsAttachable {
|
||||
// Kubelet not responsible for detaching or this volume has a non-attachable volume plugin,
|
||||
// so just remove it to actualStateOfWorld without attach.
|
||||
rc.actualStateOfWorld.MarkVolumeAsDetached(
|
||||
attachedVolume.VolumeName, rc.hostName)
|
||||
} else {
|
||||
// Only detach if kubelet detach is enabled
|
||||
glog.V(12).Infof("Attempting to start DetachVolume for volume %q (spec.Name: %q)",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name())
|
||||
err := rc.operationExecutor.DetachVolume(
|
||||
attachedVolume.AttachedVolume, false /* verifySafeToDetach */, rc.actualStateOfWorld)
|
||||
if err != nil &&
|
||||
!goroutinemap.IsAlreadyExists(err) &&
|
||||
!goroutinemap.IsExponentialBackoff(err) {
|
||||
// Ignore goroutinemap.IsAlreadyExists && goroutinemap.IsExponentialBackoff errors, they are expected.
|
||||
// Log all other errors.
|
||||
glog.Errorf(
|
||||
"operationExecutor.DetachVolume failed for volume %q (spec.Name: %q) controllerAttachDetachEnabled: %v with err: %v",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name(),
|
||||
rc.controllerAttachDetachEnabled,
|
||||
err)
|
||||
}
|
||||
if err == nil {
|
||||
glog.Infof("DetachVolume operation started for volume %q (spec.Name: %q)",
|
||||
attachedVolume.VolumeName,
|
||||
attachedVolume.VolumeSpec.Name())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
441
pkg/kubelet/volumemanager/reconciler/reconciler_test.go
Normal file
441
pkg/kubelet/volumemanager/reconciler/reconciler_test.go
Normal file
@@ -0,0 +1,441 @@
|
||||
/*
|
||||
Copyright 2016 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package reconciler
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
|
||||
"k8s.io/kubernetes/pkg/client/testing/core"
|
||||
"k8s.io/kubernetes/pkg/kubelet/volumemanager/cache"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
"k8s.io/kubernetes/pkg/util/wait"
|
||||
"k8s.io/kubernetes/pkg/volume"
|
||||
volumetesting "k8s.io/kubernetes/pkg/volume/testing"
|
||||
"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
|
||||
"k8s.io/kubernetes/pkg/volume/util/volumehelper"
|
||||
)
|
||||
|
||||
const (
|
||||
// reconcilerLoopSleepDuration is the amount of time the reconciler loop
|
||||
// waits between successive executions
|
||||
reconcilerLoopSleepDuration time.Duration = 0 * time.Millisecond
|
||||
|
||||
// waitForAttachTimeout is the maximum amount of time a
|
||||
// operationexecutor.Mount call will wait for a volume to be attached.
|
||||
waitForAttachTimeout time.Duration = 1 * time.Second
|
||||
nodeName string = "myhostname"
|
||||
)
|
||||
|
||||
// Calls Run()
|
||||
// Verifies there are no calls to attach, detach, mount, unmount, etc.
|
||||
func Test_Run_Positive_DoNothing(t *testing.T) {
|
||||
// Arrange
|
||||
volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
|
||||
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
|
||||
asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
|
||||
kubeClient := createTestClient()
|
||||
oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
|
||||
reconciler := NewReconciler(
|
||||
kubeClient,
|
||||
false, /* controllerAttachDetachEnabled */
|
||||
reconcilerLoopSleepDuration,
|
||||
waitForAttachTimeout,
|
||||
nodeName,
|
||||
dsw,
|
||||
asw,
|
||||
oex)
|
||||
|
||||
// Act
|
||||
go reconciler.Run(wait.NeverStop)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyZeroAttachCalls(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroWaitForAttachCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroMountDeviceCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroSetUpCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroTearDownCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
}
|
||||
|
||||
// Populates desiredStateOfWorld cache with one volume/pod.
|
||||
// Calls Run()
|
||||
// Verifies there is are attach/mount/etc calls and no detach/unmount calls.
|
||||
func Test_Run_Positive_VolumeAttachAndMount(t *testing.T) {
|
||||
// Arrange
|
||||
volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
|
||||
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
|
||||
asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
|
||||
kubeClient := createTestClient()
|
||||
oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
|
||||
reconciler := NewReconciler(
|
||||
kubeClient,
|
||||
false, /* controllerAttachDetachEnabled */
|
||||
reconcilerLoopSleepDuration,
|
||||
waitForAttachTimeout,
|
||||
nodeName,
|
||||
dsw,
|
||||
asw,
|
||||
oex)
|
||||
pod := &api.Pod{
|
||||
ObjectMeta: api.ObjectMeta{
|
||||
Name: "pod1",
|
||||
UID: "pod1uid",
|
||||
},
|
||||
Spec: api.PodSpec{
|
||||
Volumes: []api.Volume{
|
||||
{
|
||||
Name: "volume-name",
|
||||
VolumeSource: api.VolumeSource{
|
||||
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
|
||||
PDName: "fake-device1",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
|
||||
podName := volumehelper.GetUniquePodName(pod)
|
||||
generatedVolumeName, err := dsw.AddPodToVolume(
|
||||
podName, pod, volumeSpec, volumeSpec.Name(), "" /* volumeGidValue */)
|
||||
|
||||
// Assert
|
||||
if err != nil {
|
||||
t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
|
||||
}
|
||||
|
||||
// Act
|
||||
go reconciler.Run(wait.NeverStop)
|
||||
waitForMount(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyAttachCallCount(
|
||||
1 /* expectedAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyWaitForAttachCallCount(
|
||||
1 /* expectedWaitForAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyMountDeviceCallCount(
|
||||
1 /* expectedMountDeviceCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifySetUpCallCount(
|
||||
1 /* expectedSetUpCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroTearDownCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
}
|
||||
|
||||
// Populates desiredStateOfWorld cache with one volume/pod.
|
||||
// Enables controllerAttachDetachEnabled.
|
||||
// Calls Run()
|
||||
// Verifies there is one mount call and no unmount calls.
|
||||
// Verifies there are no attach/detach calls.
|
||||
func Test_Run_Positive_VolumeMountControllerAttachEnabled(t *testing.T) {
|
||||
// Arrange
|
||||
volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
|
||||
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
|
||||
asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
|
||||
kubeClient := createTestClient()
|
||||
oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
|
||||
reconciler := NewReconciler(
|
||||
kubeClient,
|
||||
true, /* controllerAttachDetachEnabled */
|
||||
reconcilerLoopSleepDuration,
|
||||
waitForAttachTimeout,
|
||||
nodeName,
|
||||
dsw,
|
||||
asw,
|
||||
oex)
|
||||
pod := &api.Pod{
|
||||
ObjectMeta: api.ObjectMeta{
|
||||
Name: "pod1",
|
||||
UID: "pod1uid",
|
||||
},
|
||||
Spec: api.PodSpec{
|
||||
Volumes: []api.Volume{
|
||||
{
|
||||
Name: "volume-name",
|
||||
VolumeSource: api.VolumeSource{
|
||||
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
|
||||
PDName: "fake-device1",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
|
||||
podName := volumehelper.GetUniquePodName(pod)
|
||||
generatedVolumeName, err := dsw.AddPodToVolume(
|
||||
podName, pod, volumeSpec, volumeSpec.Name(), "" /* volumeGidValue */)
|
||||
dsw.MarkVolumesReportedInUse([]api.UniqueVolumeName{generatedVolumeName})
|
||||
|
||||
// Assert
|
||||
if err != nil {
|
||||
t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
|
||||
}
|
||||
|
||||
// Act
|
||||
go reconciler.Run(wait.NeverStop)
|
||||
waitForMount(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyZeroAttachCalls(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyWaitForAttachCallCount(
|
||||
1 /* expectedWaitForAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyMountDeviceCallCount(
|
||||
1 /* expectedMountDeviceCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifySetUpCallCount(
|
||||
1 /* expectedSetUpCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroTearDownCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
}
|
||||
|
||||
// Populates desiredStateOfWorld cache with one volume/pod.
|
||||
// Calls Run()
|
||||
// Verifies there is one attach/mount/etc call and no detach calls.
|
||||
// Deletes volume/pod from desired state of world.
|
||||
// Verifies detach/unmount calls are issued.
|
||||
func Test_Run_Positive_VolumeAttachMountUnmountDetach(t *testing.T) {
|
||||
// Arrange
|
||||
volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
|
||||
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
|
||||
asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
|
||||
kubeClient := createTestClient()
|
||||
oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
|
||||
reconciler := NewReconciler(
|
||||
kubeClient,
|
||||
false, /* controllerAttachDetachEnabled */
|
||||
reconcilerLoopSleepDuration,
|
||||
waitForAttachTimeout,
|
||||
nodeName,
|
||||
dsw,
|
||||
asw,
|
||||
oex)
|
||||
pod := &api.Pod{
|
||||
ObjectMeta: api.ObjectMeta{
|
||||
Name: "pod1",
|
||||
UID: "pod1uid",
|
||||
},
|
||||
Spec: api.PodSpec{
|
||||
Volumes: []api.Volume{
|
||||
{
|
||||
Name: "volume-name",
|
||||
VolumeSource: api.VolumeSource{
|
||||
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
|
||||
PDName: "fake-device1",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
|
||||
podName := volumehelper.GetUniquePodName(pod)
|
||||
generatedVolumeName, err := dsw.AddPodToVolume(
|
||||
podName, pod, volumeSpec, volumeSpec.Name(), "" /* volumeGidValue */)
|
||||
|
||||
// Assert
|
||||
if err != nil {
|
||||
t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
|
||||
}
|
||||
|
||||
// Act
|
||||
go reconciler.Run(wait.NeverStop)
|
||||
waitForMount(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyAttachCallCount(
|
||||
1 /* expectedAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyWaitForAttachCallCount(
|
||||
1 /* expectedWaitForAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyMountDeviceCallCount(
|
||||
1 /* expectedMountDeviceCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifySetUpCallCount(
|
||||
1 /* expectedSetUpCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroTearDownCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
|
||||
// Act
|
||||
dsw.DeletePodFromVolume(podName, generatedVolumeName)
|
||||
waitForDetach(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyTearDownCallCount(
|
||||
1 /* expectedTearDownCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyDetachCallCount(
|
||||
1 /* expectedDetachCallCount */, fakePlugin))
|
||||
}
|
||||
|
||||
// Populates desiredStateOfWorld cache with one volume/pod.
|
||||
// Enables controllerAttachDetachEnabled.
|
||||
// Calls Run()
|
||||
// Verifies one mount call is made and no unmount calls.
|
||||
// Deletes volume/pod from desired state of world.
|
||||
// Verifies one unmount call is made.
|
||||
// Verifies there are no attach/detach calls made.
|
||||
func Test_Run_Positive_VolumeUnmountControllerAttachEnabled(t *testing.T) {
|
||||
// Arrange
|
||||
volumePluginMgr, fakePlugin := volumetesting.GetTestVolumePluginMgr(t)
|
||||
dsw := cache.NewDesiredStateOfWorld(volumePluginMgr)
|
||||
asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
|
||||
kubeClient := createTestClient()
|
||||
oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr)
|
||||
reconciler := NewReconciler(
|
||||
kubeClient,
|
||||
true, /* controllerAttachDetachEnabled */
|
||||
reconcilerLoopSleepDuration,
|
||||
waitForAttachTimeout,
|
||||
nodeName,
|
||||
dsw,
|
||||
asw,
|
||||
oex)
|
||||
pod := &api.Pod{
|
||||
ObjectMeta: api.ObjectMeta{
|
||||
Name: "pod1",
|
||||
UID: "pod1uid",
|
||||
},
|
||||
Spec: api.PodSpec{
|
||||
Volumes: []api.Volume{
|
||||
{
|
||||
Name: "volume-name",
|
||||
VolumeSource: api.VolumeSource{
|
||||
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{
|
||||
PDName: "fake-device1",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
volumeSpec := &volume.Spec{Volume: &pod.Spec.Volumes[0]}
|
||||
podName := volumehelper.GetUniquePodName(pod)
|
||||
generatedVolumeName, err := dsw.AddPodToVolume(
|
||||
podName, pod, volumeSpec, volumeSpec.Name(), "" /* volumeGidValue */)
|
||||
|
||||
// Assert
|
||||
if err != nil {
|
||||
t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
|
||||
}
|
||||
|
||||
// Act
|
||||
go reconciler.Run(wait.NeverStop)
|
||||
dsw.MarkVolumesReportedInUse([]api.UniqueVolumeName{generatedVolumeName})
|
||||
waitForMount(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyZeroAttachCalls(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyWaitForAttachCallCount(
|
||||
1 /* expectedWaitForAttachCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyMountDeviceCallCount(
|
||||
1 /* expectedMountDeviceCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifySetUpCallCount(
|
||||
1 /* expectedSetUpCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroTearDownCallCount(fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
|
||||
// Act
|
||||
dsw.DeletePodFromVolume(podName, generatedVolumeName)
|
||||
waitForDetach(t, fakePlugin, generatedVolumeName, asw)
|
||||
|
||||
// Assert
|
||||
assert.NoError(t, volumetesting.VerifyTearDownCallCount(
|
||||
1 /* expectedTearDownCallCount */, fakePlugin))
|
||||
assert.NoError(t, volumetesting.VerifyZeroDetachCallCount(fakePlugin))
|
||||
}
|
||||
|
||||
func waitForMount(
|
||||
t *testing.T,
|
||||
fakePlugin *volumetesting.FakeVolumePlugin,
|
||||
volumeName api.UniqueVolumeName,
|
||||
asw cache.ActualStateOfWorld) {
|
||||
err := retryWithExponentialBackOff(
|
||||
time.Duration(5*time.Millisecond),
|
||||
func() (bool, error) {
|
||||
mountedVolumes := asw.GetMountedVolumes()
|
||||
for _, mountedVolume := range mountedVolumes {
|
||||
if mountedVolume.VolumeName == volumeName {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
|
||||
return false, nil
|
||||
},
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Timed out waiting for volume %q to be attached.", volumeName)
|
||||
}
|
||||
}
|
||||
|
||||
func waitForDetach(
|
||||
t *testing.T,
|
||||
fakePlugin *volumetesting.FakeVolumePlugin,
|
||||
volumeName api.UniqueVolumeName,
|
||||
asw cache.ActualStateOfWorld) {
|
||||
err := retryWithExponentialBackOff(
|
||||
time.Duration(5*time.Millisecond),
|
||||
func() (bool, error) {
|
||||
if asw.VolumeExists(volumeName) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
},
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Timed out waiting for volume %q to be detached.", volumeName)
|
||||
}
|
||||
}
|
||||
|
||||
func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.ConditionFunc) error {
|
||||
backoff := wait.Backoff{
|
||||
Duration: initialDuration,
|
||||
Factor: 3,
|
||||
Jitter: 0,
|
||||
Steps: 6,
|
||||
}
|
||||
return wait.ExponentialBackoff(backoff, fn)
|
||||
}
|
||||
|
||||
func createTestClient() *fake.Clientset {
|
||||
fakeClient := &fake.Clientset{}
|
||||
fakeClient.AddReactor("get", "nodes",
|
||||
func(action core.Action) (bool, runtime.Object, error) {
|
||||
return true, &api.Node{
|
||||
ObjectMeta: api.ObjectMeta{Name: nodeName},
|
||||
Status: api.NodeStatus{
|
||||
VolumesAttached: []api.AttachedVolume{
|
||||
{
|
||||
Name: "fake-plugin/volume-name",
|
||||
DevicePath: "fake/path",
|
||||
},
|
||||
}},
|
||||
Spec: api.NodeSpec{ExternalID: nodeName},
|
||||
}, nil
|
||||
})
|
||||
fakeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) {
|
||||
return true, nil, fmt.Errorf("no reaction implemented for %s", action)
|
||||
})
|
||||
return fakeClient
|
||||
}
|
||||
Reference in New Issue
Block a user