Merge pull request #36922 from rkouj/refactor-operation-executor

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -124,11 +124,11 @@ func NewAttachDetachController(
 	adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
 	adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)
 	adc.attacherDetacher =
-		operationexecutor.NewOperationExecutor(
+		operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
 			kubeClient,
 			&adc.volumePluginMgr,
 			recorder,
-			false) // flag for experimental binary check for volume mount
+			false)) // flag for experimental binary check for volume mount
 	adc.nodeStatusUpdater = statusupdater.NewNodeStatusUpdater(
 		kubeClient, nodeInformer, adc.actualStateOfWorld)
 	adc.reconciler = reconciler.NewReconciler(

pkg/controller/volume/attachdetach/reconciler/reconciler_test.go

@@ -49,8 +49,7 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(
-		fakeKubeClient, volumePluginMgr, fakeRecorder, false)
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	nodeInformer := informers.NewNodeInformer(
 		fakeKubeClient, resyncPeriod)
 	nsu := statusupdater.NewNodeStatusUpdater(
@@ -81,7 +80,7 @@ func Test_Run_Positive_OneDesiredVolumeAttach(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr, fakeRecorder, false)
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, dsw, asw, ad, nsu)
@@ -127,7 +126,7 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithUnmountedVolume(t *te
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr, fakeRecorder, false)
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, dsw, asw, ad, nsu)
@@ -194,7 +193,7 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithMountedVolume(t *test
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr, fakeRecorder, false)
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, dsw, asw, ad, nsu)
@@ -261,7 +260,7 @@ func Test_Run_Negative_OneDesiredVolumeAttachThenDetachWithUnmountedVolumeUpdate
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(fakeKubeClient, volumePluginMgr, fakeRecorder, false)
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(true /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, dsw, asw, ad, nsu)

pkg/kubelet/runonce_test.go

@@ -103,7 +103,7 @@ func TestRunOnce(t *testing.T) {
 		kb.mounter,
 		kb.getPodsDir(),
 		kb.recorder,
-		false /* experimentalCheckNodeCapabilitiesBeforeMount*/)
+		false /* experimentalCheckNodeCapabilitiesBeforeMount */)
 
 	kb.networkPlugin, _ = network.InitNetworkPlugin([]network.NetworkPlugin{}, "", nettest.NewFakeHost(nil), componentconfig.HairpinNone, kb.nonMasqueradeCIDR, network.UseDefaultMTU)
 	// TODO: Factor out "StatsProvider" from Kubelet so we don't have a cyclic dependency

pkg/kubelet/volumemanager/reconciler/reconciler_test.go

@@ -60,7 +60,7 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
 	kubeClient := createTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount*/)
+	oex := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	reconciler := NewReconciler(
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
@@ -97,7 +97,7 @@ func Test_Run_Positive_VolumeAttachAndMount(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
 	kubeClient := createTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr, fakeRecorder, false)
+	oex := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	reconciler := NewReconciler(
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
@@ -168,7 +168,7 @@ func Test_Run_Positive_VolumeMountControllerAttachEnabled(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
 	kubeClient := createTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr, fakeRecorder, false)
+	oex := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	reconciler := NewReconciler(
 		kubeClient,
 		true, /* controllerAttachDetachEnabled */
@@ -240,7 +240,7 @@ func Test_Run_Positive_VolumeAttachMountUnmountDetach(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
 	kubeClient := createTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr, fakeRecorder, false)
+	oex := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	reconciler := NewReconciler(
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
@@ -323,7 +323,7 @@ func Test_Run_Positive_VolumeUnmountControllerAttachEnabled(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(nodeName, volumePluginMgr)
 	kubeClient := createTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	oex := operationexecutor.NewOperationExecutor(kubeClient, volumePluginMgr, fakeRecorder, false)
+	oex := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(kubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
 	reconciler := NewReconciler(
 		kubeClient,
 		true, /* controllerAttachDetachEnabled */

pkg/kubelet/volumemanager/volume_manager.go

@@ -164,11 +164,12 @@ func NewVolumeManager(
 		volumePluginMgr:     volumePluginMgr,
 		desiredStateOfWorld: cache.NewDesiredStateOfWorld(volumePluginMgr),
 		actualStateOfWorld:  cache.NewActualStateOfWorld(nodeName, volumePluginMgr),
-		operationExecutor: operationexecutor.NewOperationExecutor(
+		operationExecutor: operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
 			kubeClient,
 			volumePluginMgr,
 			recorder,
 			checkNodeCapabilitiesBeforeMount),
+		),
 	}
 
 	vm.reconciler = reconciler.NewReconciler(

pkg/kubelet/volumemanager/volume_manager_test.go

@@ -196,7 +196,7 @@ func newTestVolumeManager(
 		&mount.FakeMounter{},
 		"",
 		fakeRecorder,
-		false /* experimentalCheckNodeCapabilitiesBeforeMount*/)
+		false /* experimentalCheckNodeCapabilitiesBeforeMount */)
 
 	return vm, err
 }

pkg/volume/util/operationexecutor/BUILD

@@ -5,11 +5,15 @@ licenses(["notice"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
     name = "go_default_library",
-    srcs = ["operation_executor.go"],
+    srcs = [
+        "operation_executor.go",
+        "operation_generator.go",
+    ],
     tags = ["automanaged"],
     deps = [
         "//pkg/api/errors:go_default_library",
@@ -27,3 +31,16 @@ go_library(
         "//vendor:github.com/golang/glog",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["operation_executor_test.go"],
+    library = "go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//pkg/api/v1:go_default_library",
+        "//pkg/types:go_default_library",
+        "//pkg/util/mount:go_default_library",
+        "//vendor:k8s.io/client-go/_vendor/github.com/pborman/uuid",
+    ],
+)

pkg/volume/util/operationexecutor/operation_executor_test.go

+/*
+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 operationexecutor
+
+import (
+	"k8s.io/client-go/_vendor/github.com/pborman/uuid"
+	"k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util/mount"
+	"strconv"
+	"testing"
+	"time"
+)
+
+const numVolumesToMount = 2
+
+var _ OperationGenerator = &mockOperationGenerator{}
+
+func TestOperationExecutor_MountVolume_ParallelMountForNonAttachablePlugins(t *testing.T) {
+	// Arrange
+	ch, quit := make(chan interface{}), make(chan interface{})
+	numMountOperationsStarted := 0
+	mopg := newMockOperationGenerator(ch, quit)
+	oe := NewOperationExecutor(mopg)
+	volumesToMount := make([]VolumeToMount, numVolumesToMount)
+	secretName := "secret-volume"
+	volumeName := v1.UniqueVolumeName(secretName)
+
+	// Act
+	for i := range volumesToMount {
+		podName := "pod-" + strconv.Itoa((i + 1))
+		pod := getTestPodWithSecret(podName, secretName)
+		volumesToMount[i] = VolumeToMount{
+			Pod:                pod,
+			VolumeName:         volumeName,
+			PluginIsAttachable: false, // this field determines whether the plugin is attachable
+			ReportedInUse:      true,
+		}
+		oe.MountVolume(0 /* waitForAttachTimeOut */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
+	}
+
+	// Assert
+loop:
+	for {
+		select {
+		case <-ch:
+			numMountOperationsStarted++
+			if numMountOperationsStarted == numVolumesToMount {
+				break loop
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Unable to start mount operations in parallel for non-attachable volumes")
+			break loop
+		}
+	}
+	close(quit)
+}
+
+func TestOperationExecutor_MountVolume_ParallelMountForAttachablePlugins(t *testing.T) {
+	// Arrange
+	ch, quit := make(chan interface{}), make(chan interface{})
+	numMountOperationsStarted := 0
+	mopg := newMockOperationGenerator(ch, quit)
+	oe := NewOperationExecutor(mopg)
+	volumesToMount := make([]VolumeToMount, numVolumesToMount)
+	pdName := "pd-volume"
+	volumeName := v1.UniqueVolumeName(pdName)
+
+	// Act
+	for i := range volumesToMount {
+		podName := "pod-" + strconv.Itoa((i + 1))
+		pod := getTestPodWithGCEPD(podName, pdName)
+		volumesToMount[i] = VolumeToMount{
+			Pod:                pod,
+			VolumeName:         volumeName,
+			PluginIsAttachable: true, // this field determines whether the plugin is attachable
+			ReportedInUse:      true,
+		}
+		oe.MountVolume(0 /* waitForAttachTimeout */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
+	}
+
+	// Assert
+loop:
+	for {
+		select {
+		case <-ch:
+			numMountOperationsStarted++
+			if numMountOperationsStarted > 1 {
+				t.Fatalf("Mount operations should not start in parallel for attachable volumes")
+				break loop
+			}
+		case <-time.After(5 * time.Second):
+			break loop
+		}
+	}
+	close(quit)
+}
+
+type mockOperationGenerator struct {
+	ch   chan interface{}
+	quit chan interface{}
+}
+
+func newMockOperationGenerator(ch, quit chan interface{}) OperationGenerator {
+	return &mockOperationGenerator{
+		ch:   ch,
+		quit: quit,
+	}
+}
+
+func (mopg *mockOperationGenerator) GenerateMountVolumeFunc(waitForAttachTimeout time.Duration, volumeToMount VolumeToMount, actualStateOfWorldMounterUpdater ActualStateOfWorldMounterUpdater) (func() error, error) {
+	return func() error {
+		mopg.ch <- nil
+		// Blocks until the assertion is complete
+		<-mopg.quit
+		return nil
+	}, nil
+}
+func (mopg *mockOperationGenerator) GenerateUnmountVolumeFunc(volumeToUnmount MountedVolume, actualStateOfWorld ActualStateOfWorldMounterUpdater) (func() error, error) {
+	return func() error { return nil }, nil
+}
+func (mopg *mockOperationGenerator) GenerateAttachVolumeFunc(volumeToAttach VolumeToAttach, actualStateOfWorld ActualStateOfWorldAttacherUpdater) (func() error, error) {
+	return func() error { return nil }, nil
+}
+func (mopg *mockOperationGenerator) GenerateDetachVolumeFunc(volumeToDetach AttachedVolume, verifySafeToDetach bool, actualStateOfWorld ActualStateOfWorldAttacherUpdater) (func() error, error) {
+	return func() error { return nil }, nil
+}
+func (mopg *mockOperationGenerator) GenerateVolumesAreAttachedFunc(attachedVolumes []AttachedVolume, nodeName types.NodeName, actualStateOfWorld ActualStateOfWorldAttacherUpdater) (func() error, error) {
+	return func() error { return nil }, nil
+}
+func (mopg *mockOperationGenerator) GenerateUnmountDeviceFunc(deviceToDetach AttachedVolume, actualStateOfWorld ActualStateOfWorldMounterUpdater, mounter mount.Interface) (func() error, error) {
+	return func() error { return nil }, nil
+}
+func (mopg *mockOperationGenerator) GenerateVerifyControllerAttachedVolumeFunc(volumeToMount VolumeToMount, nodeName types.NodeName, actualStateOfWorld ActualStateOfWorldAttacherUpdater) (func() error, error) {
+	return func() error { return nil }, nil
+}
+
+func getTestPodWithSecret(podName, secretName string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: v1.ObjectMeta{
+			Name: podName,
+			UID:  types.UID(podName),
+		},
+		Spec: v1.PodSpec{
+			Volumes: []v1.Volume{
+				{
+					Name: secretName,
+					VolumeSource: v1.VolumeSource{
+						Secret: &v1.SecretVolumeSource{
+							SecretName: secretName,
+						},
+					},
+				},
+			},
+			Containers: []v1.Container{
+				{
+					Name:  "secret-volume-test",
+					Image: "gcr.io/google_containers/mounttest:0.7",
+					Args: []string{
+						"--file_content=/etc/secret-volume/data-1",
+						"--file_mode=/etc/secret-volume/data-1"},
+					VolumeMounts: []v1.VolumeMount{
+						{
+							Name:      secretName,
+							MountPath: "/data",
+						},
+					},
+				},
+			},
+			RestartPolicy: v1.RestartPolicyNever,
+		},
+	}
+}
+
+func getTestPodWithGCEPD(podName, pdName string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: v1.ObjectMeta{
+			Name: podName,
+			UID:  types.UID(podName + string(uuid.New())),
+		},
+		Spec: v1.PodSpec{
+			Volumes: []v1.Volume{
+				{
+					Name: pdName,
+					VolumeSource: v1.VolumeSource{
+						GCEPersistentDisk: &v1.GCEPersistentDiskVolumeSource{
+							PDName:   pdName,
+							FSType:   "ext4",
+							ReadOnly: false,
+						},
+					},
+				},
+			},
+			Containers: []v1.Container{
+				{
+					Name:  "pd-volume-test",
+					Image: "gcr.io/google_containers/mounttest:0.7",
+					Args: []string{
+						"--file_content=/etc/pd-volume/data-1",
+					},
+					VolumeMounts: []v1.VolumeMount{
+						{
+							Name:      pdName,
+							MountPath: "/data",
+						},
+					},
+				},
+			},
+			RestartPolicy: v1.RestartPolicyNever,
+		},
+	}
+}